Files
app/tests/test_capability_link.py
T

68 lines
2.5 KiB
Python

import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import AsyncMock, patch
import httpx
from starlette.applications import Starlette
from backend.shared import human_vault as vault
class CapabilityLinkTests(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.temp = tempfile.TemporaryDirectory()
self.path = Path(self.temp.name) / 'vault.sqlite'
self.env = patch.dict(os.environ, PERSIMMONS_HUMAN_VAULT_DB=str(self.path))
self.env.start()
self.actor = AsyncMock(return_value='fixture')
app = Starlette(routes=vault.routes(self.actor))
self.client = httpx.AsyncClient(
transport=httpx.ASGITransport(app=app), base_url='https://teahub.xyz',
headers={'Cookie': 'session=fixture'},
)
async def asyncTearDown(self):
await self.client.aclose()
self.env.stop()
self.temp.cleanup()
async def test_signed_in_page_mints_owner_capability_used_without_cookie(self):
response = await self.client.post(
'/chat/persimmons/human-vault/link',
headers={'Origin': 'https://persimmons.app', 'X-Requested-With': 'Persimmons'},
)
self.assertEqual(response.status_code, 200)
linked = response.json()
self.assertEqual(linked['owner'], 'fixture')
self.assertRegex(linked['capability'], vault.CAPABILITY)
self.assertNotIn(linked['capability'].encode(), self.path.read_bytes())
self.actor.reset_mock()
result = await self.client.post(
vault.PREFIX + '/read',
headers={
'Origin': vault.EXTENSION_ORIGIN,
'Cookie': '',
'X-Vault-Owner': 'fixture',
'X-Vault-Capability': linked['capability'],
},
)
self.assertEqual(result.status_code, 200)
self.assertEqual(result.json(), {'owner': 'fixture', 'items': []})
self.actor.assert_not_awaited()
async def test_link_is_exact_origin_cookie_and_request_header_gated(self):
endpoint = '/chat/persimmons/human-vault/link'
for headers in (
{'Origin': 'https://evil.example', 'X-Requested-With': 'Persimmons'},
{'Origin': 'https://persimmons.app'},
{'Origin': 'https://persimmons.app', 'X-Requested-With': 'Persimmons', 'Cookie': ''},
):
self.assertIn((await self.client.post(endpoint, headers=headers)).status_code, (401, 403))
if __name__ == '__main__':
unittest.main()