diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml
index 14ba81f..09b72cc 100644
--- a/.github/workflows/pytest.yml
+++ b/.github/workflows/pytest.yml
@@ -21,9 +21,10 @@ jobs:
- name: Install dependencies
run: |
python -m pip install --upgrade pip
- pip install flake8 pytest pytest-cov
- if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+ pip install -r requirements.txt
+ pip install -r requirements-text.txt
- name: Test with pytest
+ # Tests are intentionally covered to detect broken tests.
run: |
pytest --cov=./hippolyzer --cov=./tests --cov-report=xml
diff --git a/requirements-test.txt b/requirements-test.txt
new file mode 100644
index 0000000..a28c8f3
--- /dev/null
+++ b/requirements-test.txt
@@ -0,0 +1,4 @@
+aioresponses
+pytest
+pytest-cov
+flake8
diff --git a/setup.py b/setup.py
index 8757b19..f47b464 100644
--- a/setup.py
+++ b/setup.py
@@ -98,5 +98,6 @@ setup(
],
tests_require=[
"pytest",
+ "aioresponses",
],
)
diff --git a/tests/proxy/test_capsclient.py b/tests/proxy/test_capsclient.py
new file mode 100644
index 0000000..ef1cf43
--- /dev/null
+++ b/tests/proxy/test_capsclient.py
@@ -0,0 +1,65 @@
+import unittest
+
+import aiohttp
+import aioresponses
+from yarl import URL
+
+from hippolyzer.lib.base.datatypes import UUID
+from hippolyzer.lib.proxy.caps_client import CapsClient
+from hippolyzer.lib.proxy.region import ProxiedRegion
+from hippolyzer.lib.proxy.sessions import SessionManager
+
+
+class TestCapsClient(unittest.IsolatedAsyncioTestCase):
+ def setUp(self) -> None:
+ self.session = self.session = SessionManager().create_session({
+ "session_id": UUID.random(),
+ "secure_session_id": UUID.random(),
+ "agent_id": UUID.random(),
+ "circuit_code": 0,
+ "sim_ip": "127.0.0.1",
+ "sim_port": "1",
+ "seed_capability": "https://test.localhost:4/foo",
+ })
+ self.region = ProxiedRegion(("127.0.0.1", 1), "", self.session)
+ self.caps_client = CapsClient(self.region)
+
+ async def test_bare_url_works(self):
+ with aioresponses.aioresponses() as m:
+ m.get("https://example.com/", body=b"foo")
+ async with self.caps_client.get("https://example.com/") as resp:
+ self.assertEqual(await resp.read(), b"foo")
+
+ async def test_own_session_works(self):
+ with aioresponses.aioresponses() as m:
+ async with aiohttp.ClientSession() as sess:
+ m.get("https://example.com/", body=b"foo")
+ async with self.caps_client.get("https://example.com/", session=sess) as resp:
+ self.assertEqual(await resp.read(), b"foo")
+
+ async def test_read_llsd(self):
+ with aioresponses.aioresponses() as m:
+ m.get("https://example.com/", body=b"2")
+ async with self.caps_client.get("https://example.com/") as resp:
+ self.assertEqual(await resp.read_llsd(), 2)
+
+ async def test_caps(self):
+ self.region.update_caps({"Foobar": "https://example.com/"})
+ with aioresponses.aioresponses() as m:
+ m.post("https://example.com/baz", body=b"ok")
+ data = {"hi": "hello"}
+ headers = {"Foo": "bar"}
+ async with self.caps_client.post("Foobar", path="baz", llsd=data, headers=headers) as resp:
+ self.assertEqual(await resp.read(), b"ok")
+
+ # Our original dict should not have been touched
+ self.assertEqual(headers, {"Foo": "bar"})
+
+ req_key = ("POST", URL("https://example.com/baz"))
+ req_body = m.requests[req_key][0].kwargs['data']
+ self.assertEqual(req_body, b'')
+
+ with self.assertRaises(KeyError):
+ with self.caps_client.get("BadCap"):
+ pass