diff --git a/tests/appservice/test_api.py b/tests/appservice/test_api.py
index 15fce165b6..3c635e3dcb 100644
--- a/tests/appservice/test_api.py
+++ b/tests/appservice/test_api.py
@@ -11,18 +11,18 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from typing import Any, List, Mapping, Sequence, Union
+from typing import Any, List, Mapping, Optional, Sequence, Union
from unittest.mock import Mock
from twisted.test.proto_helpers import MemoryReactor
-from synapse.api.errors import HttpResponseException
from synapse.appservice import ApplicationService
from synapse.server import HomeServer
from synapse.types import JsonDict
from synapse.util import Clock
from tests import unittest
+from tests.unittest import override_config
PROTOCOL = "myproto"
TOKEN = "myastoken"
@@ -40,7 +40,7 @@ class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):
hs_token=TOKEN,
)
- def test_query_3pe_authenticates_token(self) -> None:
+ def test_query_3pe_authenticates_token_via_header(self) -> None:
"""
Tests that 3pe queries to the appservice are authenticated
with the appservice's token.
@@ -75,12 +75,16 @@ class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):
args: Mapping[Any, Any],
headers: Mapping[Union[str, bytes], Sequence[Union[str, bytes]]],
) -> List[JsonDict]:
- # Ensure the access token is passed as both a header and query arg.
- if not headers.get("Authorization") or not args.get(b"access_token"):
+ # Ensure the access token is passed as a header.
+ if not headers or not headers.get("Authorization"):
raise RuntimeError("Access token not provided")
+ # ... and not as a query param
+ if b"access_token" in args:
+ raise RuntimeError(
+ "Access token should not be passed as a query param."
+ )
self.assertEqual(headers.get("Authorization"), [f"Bearer {TOKEN}"])
- self.assertEqual(args.get(b"access_token"), TOKEN)
self.request_url = url
if url == URL_USER:
return SUCCESS_RESULT_USER
@@ -107,10 +111,13 @@ class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):
self.assertEqual(self.request_url, URL_LOCATION)
self.assertEqual(result, SUCCESS_RESULT_LOCATION)
- def test_fallback(self) -> None:
+ @override_config({"use_appservice_legacy_authorization": True})
+ def test_query_3pe_authenticates_token_via_param(self) -> None:
"""
- Tests that the fallback to legacy URLs works.
+ Tests that 3pe queries to the appservice are authenticated
+ with the appservice's token.
"""
+
SUCCESS_RESULT_USER = [
{
"protocol": PROTOCOL,
@@ -120,30 +127,41 @@ class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):
},
}
]
+ SUCCESS_RESULT_LOCATION = [
+ {
+ "protocol": PROTOCOL,
+ "alias": "#a:room",
+ "fields": {
+ "more": "fields",
+ },
+ }
+ ]
URL_USER = f"{URL}/_matrix/app/v1/thirdparty/user/{PROTOCOL}"
- FALLBACK_URL_USER = f"{URL}/_matrix/app/unstable/thirdparty/user/{PROTOCOL}"
+ URL_LOCATION = f"{URL}/_matrix/app/v1/thirdparty/location/{PROTOCOL}"
self.request_url = None
- self.v1_seen = False
async def get_json(
url: str,
args: Mapping[Any, Any],
- headers: Mapping[Union[str, bytes], Sequence[Union[str, bytes]]],
+ headers: Optional[
+ Mapping[Union[str, bytes], Sequence[Union[str, bytes]]]
+ ] = None,
) -> List[JsonDict]:
- # Ensure the access token is passed as both a header and query arg.
- if not headers.get("Authorization") or not args.get(b"access_token"):
- raise RuntimeError("Access token not provided")
+ # Ensure the access token is passed as a both a query param and in the headers.
+ if not args.get(b"access_token"):
+ raise RuntimeError("Access token should be provided in query params.")
+ if not headers or not headers.get("Authorization"):
+ raise RuntimeError("Access token should be provided in auth headers.")
- self.assertEqual(headers.get("Authorization"), [f"Bearer {TOKEN}"])
self.assertEqual(args.get(b"access_token"), TOKEN)
+ self.assertEqual(headers.get("Authorization"), [f"Bearer {TOKEN}"])
self.request_url = url
if url == URL_USER:
- self.v1_seen = True
- raise HttpResponseException(404, "NOT_FOUND", b"NOT_FOUND")
- elif url == FALLBACK_URL_USER:
return SUCCESS_RESULT_USER
+ elif url == URL_LOCATION:
+ return SUCCESS_RESULT_LOCATION
else:
raise RuntimeError(
"URL provided was invalid. This should never be seen."
@@ -155,9 +173,15 @@ class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):
result = self.get_success(
self.api.query_3pe(self.service, "user", PROTOCOL, {b"some": [b"field"]})
)
- self.assertTrue(self.v1_seen)
- self.assertEqual(self.request_url, FALLBACK_URL_USER)
+ 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"]}
+ )
+ )
+ self.assertEqual(self.request_url, URL_LOCATION)
+ self.assertEqual(result, SUCCESS_RESULT_LOCATION)
def test_claim_keys(self) -> None:
"""
|
