Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Commit

Permalink
Adjust tests.appservice.test_api to test both TCP and Unix Sockets
Browse files Browse the repository at this point in the history
  • Loading branch information
@realtyem
realtyem committed Oct 4, 2023
1 parent 3607771 commit 80693d5
Showing 1 changed file with 89 additions and 19 deletions.
108 changes: 89 additions & 19 deletions tests/appservice/test_api.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
from typing import Any, List, Mapping, Optional, Sequence, Union
from unittest.mock import Mock

from parameterized import parameterized

from twisted.test.proto_helpers import MemoryReactor

from synapse.appservice import ApplicationService
Expand All @@ -27,20 +29,38 @@
PROTOCOL = "myproto"
TOKEN = "myastoken"
URL = "http://mytestservice"
UNIX_URL = "unix:/var/run/testservice.socket"


class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
self.api = hs.get_application_service_api()
self.service = ApplicationService(
id="unique_identifier",
sender="@as:test",
url=URL,
token="unused",
hs_token=TOKEN,
)

def test_query_3pe_authenticates_token_via_header(self) -> None:
@parameterized.expand(
[
(
ApplicationService(
id="unique_identifier",
sender="@as:test",
url=URL,
token="unused",
hs_token=TOKEN,
),
),
(
ApplicationService(
id="unique_identifier",
sender="@as:test",
url=UNIX_URL,
token="unused",
hs_token=TOKEN,
),
),
]
)
def test_query_3pe_authenticates_token_via_header(
self, service_to_test: ApplicationService
) -> None:
"""
Tests that 3pe queries to the appservice are authenticated
with the appservice's token.
Expand All @@ -65,8 +85,10 @@ def test_query_3pe_authenticates_token_via_header(self) -> None:
}
]

URL_USER = f"{URL}/_matrix/app/v1/thirdparty/user/{PROTOCOL}"
URL_LOCATION = f"{URL}/_matrix/app/v1/thirdparty/location/{PROTOCOL}"
URL_USER = f"{service_to_test.url}/_matrix/app/v1/thirdparty/user/{PROTOCOL}"
URL_LOCATION = (
f"{service_to_test.url}/_matrix/app/v1/thirdparty/location/{PROTOCOL}"
)

self.request_url = None

Expand Down Expand Up @@ -101,20 +123,44 @@ async def get_json(
self.api.get_json = Mock(side_effect=get_json) # type: ignore[method-assign]

result = self.get_success(
self.api.query_3pe(self.service, "user", PROTOCOL, {b"some": [b"field"]})
self.api.query_3pe(service_to_test, "user", PROTOCOL, {b"some": [b"field"]})
)
self.assertEqual(self.request_url, URL_USER)
self.assertEqual(result, SUCCESS_RESULT_USER)
result = self.get_success(
self.api.query_3pe(
self.service, "location", PROTOCOL, {b"some": [b"field"]}
service_to_test, "location", PROTOCOL, {b"some": [b"field"]}
)
)
self.assertEqual(self.request_url, URL_LOCATION)
self.assertEqual(result, SUCCESS_RESULT_LOCATION)

@parameterized.expand(
[
(
ApplicationService(
id="unique_identifier",
sender="@as:test",
url=URL,
token="unused",
hs_token=TOKEN,
),
),
(
ApplicationService(
id="unique_identifier",
sender="@as:test",
url=UNIX_URL,
token="unused",
hs_token=TOKEN,
),
),
]
)
@override_config({"use_appservice_legacy_authorization": True})
def test_query_3pe_authenticates_token_via_param(self) -> None:
def test_query_3pe_authenticates_token_via_param(
self, service_to_test: ApplicationService
) -> None:
"""
Tests that 3pe queries to the appservice are authenticated
with the appservice's token.
Expand All @@ -139,8 +185,10 @@ def test_query_3pe_authenticates_token_via_param(self) -> None:
}
]

URL_USER = f"{URL}/_matrix/app/v1/thirdparty/user/{PROTOCOL}"
URL_LOCATION = f"{URL}/_matrix/app/v1/thirdparty/location/{PROTOCOL}"
URL_USER = f"{service_to_test.url}/_matrix/app/v1/thirdparty/user/{PROTOCOL}"
URL_LOCATION = (
f"{service_to_test.url}/_matrix/app/v1/thirdparty/location/{PROTOCOL}"
)

self.request_url = None

Expand Down Expand Up @@ -175,19 +223,41 @@ async def get_json(
self.api.get_json = Mock(side_effect=get_json) # type: ignore[method-assign]

result = self.get_success(
self.api.query_3pe(self.service, "user", PROTOCOL, {b"some": [b"field"]})
self.api.query_3pe(service_to_test, "user", PROTOCOL, {b"some": [b"field"]})
)
self.assertEqual(self.request_url, URL_USER)
self.assertEqual(result, SUCCESS_RESULT_USER)
result = self.get_success(
self.api.query_3pe(
self.service, "location", PROTOCOL, {b"some": [b"field"]}
service_to_test, "location", PROTOCOL, {b"some": [b"field"]}
)
)
self.assertEqual(self.request_url, URL_LOCATION)
self.assertEqual(result, SUCCESS_RESULT_LOCATION)

def test_claim_keys(self) -> None:
@parameterized.expand(
[
(
ApplicationService(
id="unique_identifier",
sender="@as:test",
url=URL,
token="unused",
hs_token=TOKEN,
),
),
(
ApplicationService(
id="unique_identifier",
sender="@as:test",
url=UNIX_URL,
token="unused",
hs_token=TOKEN,
),
),
]
)
def test_claim_keys(self, service_to_test: ApplicationService) -> None:
"""
Tests that the /keys/claim response is properly parsed for missing
keys.
Expand Down Expand Up @@ -234,7 +304,7 @@ async def post_json_get_json(

claimed_keys, missing = self.get_success(
self.api.claim_client_keys(
self.service,
service_to_test,
[
# Found devices
("@alice:example.org", "DEVICE_1", "signed_curve25519", 1),
Expand Down

0 comments on commit 80693d5

Please sign in to comment.