|
1 |
| -import logging |
2 |
| -import sys |
3 |
| -from subprocess import Popen |
4 |
| -from time import sleep |
| 1 | +import json |
5 | 2 |
|
6 | 3 | import pytest
|
7 |
| -from pretenders.client.http import HTTPMock |
8 |
| -from pretenders.common.constants import FOREVER |
9 | 4 |
|
10 | 5 | import ixmp
|
11 | 6 | from ixmp.testing import create_test_platform
|
12 | 7 |
|
13 |
| -log = logging.getLogger(__name__) |
14 | 8 |
|
| 9 | +@pytest.fixture |
| 10 | +def mock(httpserver): |
| 11 | + """Mock server with responses for both tests.""" |
| 12 | + from werkzeug import Request, Response |
15 | 13 |
|
16 |
| -@pytest.fixture(scope="session") |
17 |
| -def server(): |
18 |
| - proc = Popen( |
19 |
| - [ |
20 |
| - sys.executable, |
21 |
| - "-m", |
22 |
| - "pretenders.server.server", |
23 |
| - "--host", |
24 |
| - "localhost", |
25 |
| - "--port", |
26 |
| - "8000", |
27 |
| - ] |
| 14 | + httpserver.expect_request("/login", method="POST").respond_with_json( |
| 15 | + "security-token" |
28 | 16 | )
|
29 |
| - log.info(f"Mock server started with pid {proc.pid}") |
30 |
| - |
31 |
| - # Wait for server to start up |
32 |
| - sleep(5) |
33 | 17 |
|
34 |
| - yield |
| 18 | + # Mock the behaviour of the ixmp_source (Java) access API |
| 19 | + # - Request data is valid JSON containing a list of dict. |
| 20 | + # - Response is a JSON list of bool of equal length. |
| 21 | + def handler(r: Request) -> Response: |
| 22 | + data = r.get_json() |
| 23 | + result = [ |
| 24 | + (i["username"], i["entityType"], i["entityId"]) |
| 25 | + == ("test_user", "MODEL", "test_model") |
| 26 | + for i in data |
| 27 | + ] |
| 28 | + return Response(json.dumps(result), content_type="application/json") |
35 | 29 |
|
36 |
| - proc.terminate() |
37 |
| - log.info("Mock server terminated") |
| 30 | + # Use the same handler for all test requests against the /access/list URL |
| 31 | + httpserver.expect_request( |
| 32 | + "/access/list", |
| 33 | + method="POST", |
| 34 | + headers={"Authorization": "Bearer security-token"}, # JSON Web Token header |
| 35 | + ).respond_with_handler(handler) |
38 | 36 |
|
| 37 | + return httpserver |
39 | 38 |
|
40 |
| -@pytest.fixture(scope="function") |
41 |
| -def mock(server): |
42 |
| - # Create the mock server |
43 |
| - httpmock = HTTPMock("localhost", 8000) |
44 | 39 |
|
45 |
| - # Common responses for both tests |
46 |
| - httpmock.when("POST /login").reply( |
47 |
| - '"security-token"', headers={"Content-Type": "application/json"}, times=FOREVER |
| 40 | +@pytest.fixture |
| 41 | +def test_props(mock, request, tmp_path, test_data_path): |
| 42 | + return create_test_platform( |
| 43 | + tmp_path, test_data_path, "test_access", auth_url=mock.url_for("") |
48 | 44 | )
|
49 | 45 |
|
50 |
| - yield httpmock |
51 | 46 |
|
| 47 | +M = ["test_model", "non_existing_model"] |
52 | 48 |
|
53 |
| -def test_check_single_model_access(mock, tmp_path, test_data_path, request): |
54 |
| - mock.when( |
55 |
| - "POST /access/list", |
56 |
| - body='.+"test_user".+', |
57 |
| - headers={"Authorization": "Bearer security-token"}, |
58 |
| - ).reply("[true]", headers={"Content-Type": "application/json"}, times=FOREVER) |
59 |
| - mock.when( |
60 |
| - "POST /access/list", |
61 |
| - body='.+"non_granted_user".+', |
62 |
| - headers={"Authorization": "Bearer security-token"}, |
63 |
| - ).reply("[false]", headers={"Content-Type": "application/json"}, times=FOREVER) |
64 |
| - |
65 |
| - test_props = create_test_platform( |
66 |
| - tmp_path, |
67 |
| - test_data_path, |
68 |
| - f"{request.node.name}", |
69 |
| - auth_url=mock.pretend_url, |
70 |
| - ) |
71 | 49 |
|
| 50 | +@pytest.mark.parametrize( |
| 51 | + "user, models, exp", |
| 52 | + ( |
| 53 | + ("test_user", "test_model", True), |
| 54 | + ("non_granted_user", "test_model", False), |
| 55 | + ("non_existing_user", "test_model", False), |
| 56 | + ("test_user", M, {"test_model": True, "non_existing_model": False}), |
| 57 | + ("non_granted_user", M, {"test_model": False, "non_existing_model": False}), |
| 58 | + ("non_existing_user", M, {"test_model": False, "non_existing_model": False}), |
| 59 | + ), |
| 60 | +) |
| 61 | +def test_check_access(test_props, user, models, exp): |
| 62 | + """:meth:`.check_access` correctly handles certain arguments and responses.""" |
72 | 63 | mp = ixmp.Platform(backend="jdbc", dbprops=test_props)
|
73 |
| - |
74 |
| - granted = mp.check_access("test_user", "test_model") |
75 |
| - assert granted |
76 |
| - |
77 |
| - granted = mp.check_access("non_granted_user", "test_model") |
78 |
| - assert not granted |
79 |
| - |
80 |
| - granted = mp.check_access("non_existing_user", "test_model") |
81 |
| - assert not granted |
82 |
| - |
83 |
| - |
84 |
| -def test_check_multi_model_access(mock, tmp_path, test_data_path, request): |
85 |
| - mock.when( |
86 |
| - "POST /access/list", |
87 |
| - body='.+"test_user".+', |
88 |
| - headers={"Authorization": "Bearer security-token"}, |
89 |
| - ).reply( |
90 |
| - "[true, false]", headers={"Content-Type": "application/json"}, times=FOREVER |
91 |
| - ) |
92 |
| - mock.when( |
93 |
| - "POST /access/list", |
94 |
| - body='.+"non_granted_user".+', |
95 |
| - headers={"Authorization": "Bearer security-token"}, |
96 |
| - ).reply( |
97 |
| - "[false, false]", headers={"Content-Type": "application/json"}, times=FOREVER |
98 |
| - ) |
99 |
| - |
100 |
| - test_props = create_test_platform( |
101 |
| - tmp_path, test_data_path, f"{request.node.name}", auth_url=mock.pretend_url |
102 |
| - ) |
103 |
| - |
104 |
| - mp = ixmp.Platform(backend="jdbc", dbprops=test_props) |
105 |
| - |
106 |
| - access = mp.check_access("test_user", ["test_model", "non_existing_model"]) |
107 |
| - assert access["test_model"] |
108 |
| - assert not access["non_existing_model"] |
109 |
| - |
110 |
| - access = mp.check_access("non_granted_user", ["test_model", "non_existing_model"]) |
111 |
| - assert not access["test_model"] |
112 |
| - assert not access["non_existing_model"] |
113 |
| - |
114 |
| - access = mp.check_access("non_existing_user", ["test_model", "non_existing_model"]) |
115 |
| - assert not access["test_model"] |
116 |
| - assert not access["non_existing_model"] |
| 64 | + assert exp == mp.check_access(user, models) |
0 commit comments