|
16 | 16 | ) |
17 | 17 |
|
18 | 18 |
|
19 | | -# def test_collections_filter_contained_by_token( |
20 | | -# mock_upstream, source_api_server, token_builder |
21 | | -# ): |
22 | | -# """Test that the collections filter is applied correctly.""" |
23 | | -# app = app_factory( |
24 | | -# upstream_url=source_api_server, |
25 | | -# collections_filter={ |
26 | | -# "cls": "stac_auth_proxy.filters.Template", |
27 | | -# "args": [ |
28 | | -# "A_CONTAINEDBY(id, ('{{ token.collections | join(\"', '\") }}' ))" |
29 | | -# ], |
30 | | -# }, |
31 | | -# ) |
32 | | - |
33 | | -# auth_token = token_builder({"collections": ["foo", "bar"]}) |
34 | | -# client = TestClient(app, headers={"Authorization": f"Bearer {auth_token}"}) |
35 | | -# response = client.get("/collections") |
36 | | - |
37 | | -# assert response.status_code == 200 |
38 | | -# assert mock_upstream.call_count == 1 |
39 | | -# [r] = mock_upstream.call_args[0] |
40 | | -# assert parse_qs(r.url.query.decode()) == { |
41 | | -# "filter": ["a_containedby(id, ('foo', 'bar'))"] |
42 | | -# } |
43 | | - |
44 | | - |
45 | | -# @pytest.mark.parametrize( |
46 | | -# "authenticated, expected_filter", |
47 | | -# [ |
48 | | -# (True, "true"), |
49 | | -# (False, "(private = false)"), |
50 | | -# ], |
51 | | -# ) |
52 | | -# def test_collections_filter_private_and_public( |
53 | | -# mock_upstream, source_api_server, token_builder, authenticated, expected_filter |
54 | | -# ): |
55 | | -# """Test that filter can be used for private/public collections.""" |
56 | | -# app = app_factory( |
57 | | -# upstream_url=source_api_server, |
58 | | -# collections_filter={ |
59 | | -# "cls": "stac_auth_proxy.filters.Template", |
60 | | -# "args": ["{{ '(private = false)' if token is none else true }}"], |
61 | | -# }, |
62 | | -# default_public=True, |
63 | | -# ) |
64 | | - |
65 | | -# client = TestClient( |
66 | | -# app, |
67 | | -# headers=( |
68 | | -# {"Authorization": f"Bearer {token_builder({})}"} if authenticated else {} |
69 | | -# ), |
70 | | -# ) |
71 | | -# response = client.get("/collections") |
72 | | - |
73 | | -# assert response.status_code == 200 |
74 | | -# assert mock_upstream.call_count == 1 |
75 | | -# [r] = mock_upstream.call_args[0] |
76 | | -# assert parse_qs(r.url.query.decode()) == {"filter": [expected_filter]} |
77 | | - |
78 | | - |
79 | | -# @pytest.mark.parametrize( |
80 | | -# "authenticated, expected_filter", |
81 | | -# [ |
82 | | -# (True, "true"), |
83 | | -# (False, '("properties.private" = false)'), |
84 | | -# ], |
85 | | -# ) |
86 | | -# def test_items_filter_private_and_public( |
87 | | -# mock_upstream, source_api_server, token_builder, authenticated, expected_filter |
88 | | -# ): |
89 | | -# """Test that filter can be used for private/public collections.""" |
90 | | -# app = app_factory( |
91 | | -# upstream_url=source_api_server, |
92 | | -# items_filter={ |
93 | | -# "cls": "stac_auth_proxy.filters.Template", |
94 | | -# "args": ["{{ '(properties.private = false)' if token is none else true }}"], |
95 | | -# }, |
96 | | -# default_public=True, |
97 | | -# ) |
98 | | - |
99 | | -# client = TestClient( |
100 | | -# app, |
101 | | -# headers=( |
102 | | -# {"Authorization": f"Bearer {token_builder({})}"} if authenticated else {} |
103 | | -# ), |
104 | | -# ) |
105 | | -# response = client.get("/collections/foo/items") |
106 | | - |
107 | | -# assert response.status_code == 200 |
108 | | -# assert mock_upstream.call_count == 1 |
109 | | -# [r] = mock_upstream.call_args[0] |
110 | | -# assert parse_qs(r.url.query.decode()) == {"filter": [expected_filter]} |
111 | | - |
112 | | - |
113 | | -# @pytest.mark.parametrize( |
114 | | -# "filter_template_expr, user_search, expected_anon_search, expected_authenticated_search", |
115 | | -# [ |
116 | | -# ( |
117 | | -# "{{ '(properties.private = false)' if token is none else true }}", |
118 | | -# {}, |
119 | | -# {"filter": "(properties.private = false)"}, |
120 | | -# {}, |
121 | | -# ), |
122 | | -# # ({}, {"filter": {"op": "=", "args": [{"property": "private"}, False]}}, {}), |
123 | | -# # (True, {"op": "=", "args": [{"property": "private"}, True]}), |
124 | | -# ], |
125 | | -# ) |
126 | | -# def test_search_filters( |
127 | | -# mock_upstream, |
128 | | -# source_api_server, |
129 | | -# token_builder, |
130 | | -# filter_template_expr, |
131 | | -# user_search, |
132 | | -# expected_anon_search, |
133 | | -# expected_authenticated_search, |
134 | | -# ): |
135 | | -# """Test that filter can be used for private/public collections.""" |
136 | | -# app = app_factory( |
137 | | -# upstream_url=source_api_server, |
138 | | -# items_filter={ |
139 | | -# "cls": "stac_auth_proxy.filters.Template", |
140 | | -# "args": [filter_template_expr], |
141 | | -# }, |
142 | | -# default_public=True, |
143 | | -# ) |
144 | | - |
145 | | -# for is_authenticated, expected_search in [ |
146 | | -# (True, expected_authenticated_search), |
147 | | -# (False, expected_anon_search), |
148 | | -# ]: |
149 | | -# headers = ( |
150 | | -# {"Authorization": f"Bearer {token_builder({})}"} if is_authenticated else {} |
151 | | -# ) |
152 | | -# client = TestClient(app, headers=headers) |
153 | | -# response = client.post("/search", json=user_search) |
154 | | - |
155 | | -# assert response.status_code == 200 |
156 | | -# assert mock_upstream.call_count == 1 |
157 | | -# [r] = cast(list[Request], mock_upstream.call_args[0]) |
158 | | -# body = r.read().decode() |
159 | | -# print(f"req {json.loads(body)=}") |
160 | | -# assert json.loads(body) == expected_search |
161 | | -# mock_upstream.reset_mock() |
162 | | - |
163 | | - |
164 | 19 | @pytest.mark.parametrize( |
165 | 20 | "filter_template_expr, auth_filter, anon_filter", |
166 | 21 | [ |
|
0 commit comments