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

Add authentication to thirdparty bridge APIs #12746

Merged
merged 11 commits into from
May 24, 2022
Prev Previous commit
Next Next commit
Tidy tidy according to review
  • Loading branch information
Half-Shot committed May 18, 2022
commit 512ebcaaea4a47fa5ef5b3049316cf34fff7014e
66 changes: 34 additions & 32 deletions tests/appservice/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,42 +27,52 @@
PROTOCOL = "myproto"
TOKEN = "myastoken"
URL = "http://mytestservice"
URL_USER = f"{URL}/_matrix/app/unstable/thirdparty/user/{PROTOCOL}"
URL_LOCATION = f"{URL}/_matrix/app/unstable/thirdparty/location/{PROTOCOL}"
SUCCESS_RESULT_USER = [
{
"protocol": PROTOCOL,
"userid": "@a:user",
"fields": {
"more": "fields",
},
}
]
SUCCESS_RESULT_LOCATION = [
{
"protocol": PROTOCOL,
"alias": "#a:room",
"fields": {
"more": "fields",
},
}
]


class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer):
self.api = ApplicationServiceApi(hs)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
self.api = ApplicationServiceApi(hs)
self.api = hs.get_application_service_api()

It probably doesn't make that much of a difference, but it's more consistent with how we access this kind of objects in tests.

self.service = ApplicationService(
id="unique_identifier",
sender="@as:test",
url=URL,
token="unused",
hs_token=TOKEN,
hostname="myserver",
)

def test_query_3pe_authenticates_token(self):

SUCCESS_RESULT_USER = [
{
"protocol": PROTOCOL,
"userid": "@a:user",
"fields": {
"more": "fields",
},
}
]
SUCCESS_RESULT_LOCATION = [
{
"protocol": PROTOCOL,
"alias": "#a:room",
"fields": {
"more": "fields",
},
}
]

URL_USER = f"{URL}/_matrix/app/unstable/thirdparty/user/{PROTOCOL}"
URL_LOCATION = f"{URL}/_matrix/app/unstable/thirdparty/location/{PROTOCOL}"

self.request_url = None
self.fields = None

async def get_json(url: str, args: Mapping[Any, Any]) -> List[JsonDict]:
if not args.get(b"access_token"):
raise Exception("Access token not provided")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
raise Exception("Access token not provided")
raise RuntimeError("Access token not provided")

For consistency


self.assertEqual(args.get(b"access_token"), TOKEN)
self.request_url = url
self.fields = args
if url == URL_USER:
return SUCCESS_RESULT_USER
elif url == URL_LOCATION:
Expand All @@ -71,17 +81,9 @@ async def get_json(url: str, args: Mapping[Any, Any]) -> List[JsonDict]:
self.fail("URL provided was invalid")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note that this is now a bit unnecessary since raising the RuntimeError will automatically fail the test.

return []

self.api.get_json = Mock(side_effect=get_json) # type: ignore[assignment] # We assign to a method.
self.service = ApplicationService(
id="unique_identifier",
sender="@as:test",
url=URL,
token="unused",
hs_token=TOKEN,
hostname="myserver",
)
# We assign to a method, which mypy doesn't like.
self.api.get_json = Mock(side_effect=get_json) # type: ignore[assignment]

def test_query_3pe_authenticates_token(self):
result = self.get_success(
self.api.query_3pe(self.service, "user", PROTOCOL, {b"some": [b"field"]})
)
Expand Down