afterhours-shift-manager/tests/shared/test_three_cx_client.py
Adam Moussa 3a26343cb7
Add pytest suite and wire it into CI (#85) (#86)
* Add pytest suite and wire it into CI

Stands up the first automated tests for the repo (151 tests) and turns on
the CI test step.

- Lift slack-bot handlers out of create_app() closures to module level so
  they're unit-testable; create_app is now a thin Bolt-wiring layer. No
  behavior change (handler entrypoints and create_app signature unchanged).
- tests/ mirrors src/: shared layer (schedule, blocks, 3CX client,
  ring_scheduler, secrets) + all four Lambdas (pay math, drop/swap/pick/
  admin/register/rate, pickup button, roster sync, queue scheduler).
- All boundaries mocked: DynamoDB/SES/Secrets via moto, 3CX HTTP via
  responses, Slack via fakes, time via freezegun. No real network/AWS.
- pyproject.toml pytest config (pythonpath=src/shared, importlib mode);
  per-package conftest loads each app.py under a unique name to avoid the
  four-app.py collision. tests/requirements.txt for test-only deps.
- ci.yaml: run-tests: true (reusable workflow auto-installs deps) and lint
  the tests dir too.
- README Testing section.

Closes #85

* Add least-privilege permissions block to CI workflow

Resolves the CodeQL actions/missing-workflow-permissions alert: the CI
workflow now restricts GITHUB_TOKEN to contents: read (it only checks out,
lints, and runs tests).

* Stop logging extension numbers in 3CX queue updates

Resolves 3 high CodeQL py/clear-text-logging-sensitive-data alerts: the
queue/ring-group forwarding logs no longer include the routed extension
values (closed/holiday/extension). Non-sensitive context (resource id,
queue number) is retained.
2026-06-01 19:07:08 -04:00

113 lines
3.3 KiB
Python

"""Tests for shared.three_cx_client — auth flows and XAPI calls (HTTP mocked)."""
import responses
from shared.three_cx_client import ThreeCXClient
BASE = "https://test.3cx.us"
def _stub_oauth():
responses.add(
responses.POST,
f"{BASE}/connect/token",
json={"access_token": "tok-oauth"},
status=200,
)
@responses.activate
def test_oauth_authentication_sets_bearer_header():
_stub_oauth()
client = ThreeCXClient(
domain="test.3cx.us",
auth_mode="oauth",
client_id="cid",
client_secret="secret",
)
assert client.session.headers["Authorization"] == "Bearer tok-oauth"
@responses.activate
def test_user_authentication_extracts_nested_token():
responses.add(
responses.POST,
f"{BASE}/webclient/api/Login/GetAccessToken",
json={"Token": {"access_token": "tok-user"}},
status=200,
)
client = ThreeCXClient(domain="test.3cx.us", username="u", password="p")
assert client.session.headers["Authorization"] == "Bearer tok-user"
@responses.activate
def test_user_authentication_missing_token_raises():
import pytest
responses.add(
responses.POST,
f"{BASE}/webclient/api/Login/GetAccessToken",
json={"nope": True},
status=200,
)
with pytest.raises(ValueError):
ThreeCXClient(domain="test.3cx.us", username="u", password="p")
@responses.activate
def test_get_group_members_resolves_group_then_members():
_stub_oauth()
responses.add(
responses.GET,
f"{BASE}/xapi/v1/Groups",
json={"value": [{"Id": 5, "Name": "DEFAULT"}]},
status=200,
)
responses.add(
responses.GET,
f"{BASE}/xapi/v1/Groups(5)/Members",
json={"value": [{"Number": "114", "MemberName": "Alice", "Type": "Extension"}]},
status=200,
)
client = ThreeCXClient(
domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
)
members = client.get_group_members("DEFAULT")
assert members == [{"Number": "114", "MemberName": "Alice", "Type": "Extension"}]
@responses.activate
def test_get_group_members_unknown_group_returns_empty():
_stub_oauth()
responses.add(
responses.GET, f"{BASE}/xapi/v1/Groups", json={"value": []}, status=200
)
client = ThreeCXClient(
domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
)
assert client.get_group_members("NOPE") == []
@responses.activate
def test_get_queue_and_update_queue_forwarding():
_stub_oauth()
responses.add(
responses.GET,
f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='800')",
json={"Id": 7, "Number": "800"},
status=200,
)
patched = responses.add(responses.PATCH, f"{BASE}/xapi/v1/Queues(7)", status=200)
client = ThreeCXClient(
domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
)
queue = client.get_queue("800")
assert queue["Id"] == 7
status = client.update_queue_forwarding(queue_id=7, closed="114", holiday="114")
assert status == 200
# The forwarding payload routes both closed and holiday to the extension.
import json
body = json.loads(patched.calls[0].request.body)
assert body["OutOfOfficeRoute"]["Route"]["Number"] == "114"
assert body["HolidaysRoute"]["Route"]["Number"] == "114"