FilesExpand file tree

main

/

test_legacy_parameters.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

135 lines (100 loc) · 4.5 KB

main

/

test_legacy_parameters.py

Copy path

Top

File metadata and controls

• Code
• Blame

135 lines (100 loc) · 4.5 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

# Copyright (c) Microsoft Corporation. All rights reserved.

"""Existing callers keep their construction style when an RPC record gains fields."""

import dataclasses

import pytest

from copilot.generated import rpc

def _contract() -> rpc.CatalogClientContract:

return rpc.CatalogClientContract(

protocol_version=3, required_capabilities=["mcp-install-planning"]

)

def _candidate() -> rpc.MCPPlanInstallSourceCandidate:

return rpc.MCPPlanInstallSourceCandidate(candidate_handle="candidate", search_id="search")

def _positional(cls: type) -> list[str]:

return [field.name for field in dataclasses.fields(cls) if not field.kw_only]

def test_plan_install_positional_scope_still_binds_scope():

request = rpc.MCPPlanInstallRequest(_contract(), _candidate(), rpc.MCPPlanScope.USER)

assert request.scope is rpc.MCPPlanScope.USER

assert request.policy_session_id is None

assert "policySessionId" not in request.to_dict()

assert request.to_dict()["scope"] == "user"

def test_added_fields_are_keyword_only():

assert _positional(rpc.MCPPlanInstallRequest) == ["contract", "source", "scope"]

assert _positional(rpc.CatalogSearchRequest) == ["contract", "query", "kinds", "limit", "page"]

with pytest.raises(TypeError):

rpc.MCPPlanInstallRequest(_contract(), _candidate(), rpc.MCPPlanScope.USER, "session")

def test_search_positional_arguments_keep_their_meaning():

request = rpc.CatalogSearchRequest(_contract(), "catalogue query", None, 4)

assert request.limit == 4

assert request.policy_session_id is None

assert "policySessionId" not in request.to_dict()

def test_new_fields_round_trip_by_keyword():

plan = rpc.MCPPlanInstallRequest(

_contract(), _candidate(), rpc.MCPPlanScope.USER, policy_session_id="session"

)

search = rpc.CatalogSearchRequest(_contract(), "catalogue query", policy_session_id="session")

assert rpc.MCPPlanInstallRequest.from_dict(plan.to_dict()) == plan

assert rpc.CatalogSearchRequest.from_dict(search.to_dict()) == search

assert plan.to_dict()["policySessionId"] == "session"

assert search.to_dict()["policySessionId"] == "session"

def test_existing_action_export_keeps_its_name():

assert "Action" in rpc.__all__

assert {member.value for member in rpc.Action} == {"preserve", "transform"}

assert rpc.ProtocolMarkerSectionOverride.__dataclass_fields__["action"].type in (

rpc.Action,

"Action",

)

def test_listed_server_positional_arguments_keep_their_meaning():

metadata = rpc.McpServerMetadata(instructions="use it")

server = rpc.MCPServer("server", rpc.McpServerStatus.CONNECTED, "Server", "failed", metadata)

assert server.server_metadata == metadata

assert server.owned is None

assert "owned" not in server.to_dict()

assert _positional(rpc.MCPServer) == [

"name",

"status",

"display_name",

"error",

"server_metadata",

"source",

"source_plugin",

"source_plugin_version",

]

def test_listed_server_owned_marker_round_trips_by_keyword():

server = rpc.MCPServer.from_dict(

{"name": "server", "status": "stopped", "owned": {"installationId": "installation"}}

)

assert server.owned.installation_id == "installation"

assert rpc.MCPServer.from_dict(server.to_dict()) == server

assert server.to_dict()["owned"] == {"installationId": "installation"}

def test_workspace_diff_positional_arguments_keep_their_meaning():

change = rpc.WorkspaceDiffFileChange(

rpc.WorkspaceDiffFileChangeType.MODIFIED, "patch", "app.txt", True, "old.txt"

)

assert (change.is_truncated, change.old_path, change.contents, _positional(type(change))) == (

True,

"old.txt",

None,

["change_type", "diff", "path", "is_truncated", "old_path"],

)

assert "contents" not in change.to_dict()

def test_workspace_diff_contents_round_trip_by_keyword():

contents = rpc.WorkspaceDiffContents(before="port=3000\n", after="port=4000\n")

change = rpc.WorkspaceDiffFileChange(

rpc.WorkspaceDiffFileChangeType.MODIFIED,

"patch",

"app.txt",

False,

None,

contents=contents,

)

encoded = change.to_dict()

assert (encoded, rpc.WorkspaceDiffFileChange.from_dict(encoded)) == (

{

"changeType": "modified",

"diff": "patch",

"path": "app.txt",

"isTruncated": False,

"contents": {"before": "port=3000\n", "after": "port=4000\n"},

},

change,

)