Source code
Revision control
Copy as Markdown
Other Tools
Test Info:
- Manifest: tools/tryselect/test/python.toml
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
import json
import os
import threading
import time
import urllib.parse
import urllib.request
import mozunit
import pytest
from tryselect.util.taskcluster import (
TC_ROOT_URL,
_scopes_key,
get_client,
)
import taskcluster as tc_module
DEFAULT_SCOPES = ["some:scope"]
BROWSER_CLIENT_ID = "browser-client"
BROWSER_ACCESS_TOKEN = "browser-token"
@pytest.fixture
def credentials_file(tmp_path, monkeypatch):
creds_path = tmp_path / "tc_credentials.json"
monkeypatch.setattr(
"tryselect.util.taskcluster._get_credentials_file", lambda: creds_path
)
return creds_path
def make_cache(credentials_file, scopes=None, expires_offset=7200):
scopes = scopes or DEFAULT_SCOPES
credentials_file.write_text(
json.dumps({
_scopes_key(scopes): {
"clientId": "cached-client",
"accessToken": "cached-token",
"expires": time.time() + expires_offset,
}
})
)
@pytest.fixture
def run_get_client(monkeypatch):
def fake_webbrowser_open(url):
params = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
callback_url = params["callback_url"][0]
qs = urllib.parse.urlencode({
"clientId": BROWSER_CLIENT_ID,
"accessToken": BROWSER_ACCESS_TOKEN,
})
def send_creds():
try:
urllib.request.urlopen(f"{callback_url}?{qs}", timeout=10)
except Exception:
pass
threading.Thread(target=send_creds, daemon=True).start()
def inner(service="queue", scopes=None, env=None):
scopes = scopes or DEFAULT_SCOPES
env = env or {}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("webbrowser.open", fake_webbrowser_open)
return get_client(service, scopes)
return inner
def test_get_client_automation(run_get_client):
result = run_get_client(
env={
"MOZ_AUTOMATION": "1",
"TASKCLUSTER_CLIENT_ID": "env-client",
"TASKCLUSTER_ACCESS_TOKEN": "env-token",
},
)
assert isinstance(result, tc_module.Queue)
assert result.options["credentials"]["clientId"] == b"env-client"
assert result.options["credentials"]["accessToken"] == b"env-token"
def test_get_client_cache_hit(credentials_file, run_get_client):
make_cache(credentials_file)
result = run_get_client()
assert isinstance(result, tc_module.Queue)
assert result.options["rootUrl"] == TC_ROOT_URL
assert result.options["credentials"]["clientId"] == b"cached-client"
assert result.options["credentials"]["accessToken"] == b"cached-token"
def test_get_client_cache_expired(credentials_file, run_get_client):
make_cache(credentials_file, expires_offset=200)
result = run_get_client()
assert isinstance(result, tc_module.Queue)
assert result.options["rootUrl"] == TC_ROOT_URL
assert result.options["credentials"]["clientId"] == b"browser-client"
assert result.options["credentials"]["accessToken"] == b"browser-token"
def test_get_client_browser_auth(credentials_file, run_get_client):
assert not credentials_file.exists()
result = run_get_client()
assert isinstance(result, tc_module.Queue)
assert result.options["rootUrl"] == TC_ROOT_URL
assert result.options["credentials"]["clientId"] == b"browser-client"
assert result.options["credentials"]["accessToken"] == b"browser-token"
assert credentials_file.is_file()
if __name__ == "__main__":
mozunit.main()