@@ 67-92 (lines=26) @@ | ||
64 | assert flow["switch"] == dpid |
|
65 | ||
66 | ||
67 | async def test_get_stored_flows_no_cookies_filter( |
|
68 | monkeypatch, intra_evc_evpl_flows_data |
|
69 | ) -> None: |
|
70 | """Test get_stored_flows no cookies.""" |
|
71 | evc_data = intra_evc_evpl_flows_data |
|
72 | dpid = "00:00:00:00:00:00:00:01" |
|
73 | cookies = [evc_data[dpid][0]["flow"]["cookie"]] |
|
74 | ||
75 | aclient_mock, awith_mock = AsyncMock(), MagicMock() |
|
76 | aclient_mock.get.return_value = Response( |
|
77 | 200, json=intra_evc_evpl_flows_data, request=MagicMock() |
|
78 | ) |
|
79 | awith_mock.return_value.__aenter__.return_value = aclient_mock |
|
80 | monkeypatch.setattr("httpx.AsyncClient", awith_mock) |
|
81 | ||
82 | data = await get_stored_flows() |
|
83 | assert ( |
|
84 | aclient_mock.get.call_args[0][0] == "/stored_flows?" |
|
85 | "state=installed&state=pending" |
|
86 | ) |
|
87 | assert len(data) == 1 |
|
88 | assert list(data.keys()) == cookies |
|
89 | assert len(data[cookies[0]]) == 2 |
|
90 | for flows in data.values(): |
|
91 | for flow in flows: |
|
92 | assert flow["switch"] == dpid |
|
93 | ||
94 | ||
95 | async def test_add_evcs_metadata(monkeypatch): |
|
@@ 41-64 (lines=24) @@ | ||
38 | assert data[evc_id] == evc_data |
|
39 | ||
40 | ||
41 | async def test_get_stored_flows(monkeypatch, intra_evc_evpl_flows_data) -> None: |
|
42 | """Test get_stored_flows.""" |
|
43 | evc_data = intra_evc_evpl_flows_data |
|
44 | dpid = "00:00:00:00:00:00:00:01" |
|
45 | cookies = [evc_data[dpid][0]["flow"]["cookie"]] |
|
46 | ||
47 | aclient_mock, awith_mock = AsyncMock(), MagicMock() |
|
48 | aclient_mock.request.return_value = Response( |
|
49 | 200, json=intra_evc_evpl_flows_data, request=MagicMock() |
|
50 | ) |
|
51 | awith_mock.return_value.__aenter__.return_value = aclient_mock |
|
52 | monkeypatch.setattr("httpx.AsyncClient", awith_mock) |
|
53 | ||
54 | data = await get_stored_flows(cookies) |
|
55 | assert ( |
|
56 | aclient_mock.request.call_args[0][1] == "/stored_flows?" |
|
57 | "state=installed&state=pending" |
|
58 | ) |
|
59 | assert len(data) == 1 |
|
60 | assert list(data.keys()) == cookies |
|
61 | assert len(data[cookies[0]]) == 2 |
|
62 | for flows in data.values(): |
|
63 | for flow in flows: |
|
64 | assert flow["switch"] == dpid |
|
65 | ||
66 | ||
67 | async def test_get_stored_flows_no_cookies_filter( |