85e019783b
- Introduced tests for case creation commands, ensuring proper handling of UUIDs and payloads. - Added tests for v5 assignment functionality, validating employee identification and group assignment. - Implemented tests for remote recovery scenarios, including handling network failures and response validation. - Created tests for hardware label generation and validation, ensuring proper URL formatting and QR code generation. - Developed extensive tests for remote control functionality, covering connection isolation, command execution, and user permissions. - Added tests for workflow timer registration, ensuring correct handling of timer states and ownership validation.
470 lines
26 KiB
Python
470 lines
26 KiB
Python
"""Security and real PostgreSQL remote tests in a disposable isolated schema.
|
|
|
|
Run integration cases with BMC_REMOTE_DB_TESTS=1; only localhost is accepted.
|
|
No production schema, customer record or existing time entry is modified.
|
|
"""
|
|
import asyncio
|
|
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
|
|
from datetime import datetime, timezone, timedelta
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
from urllib.parse import urlsplit
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.testclient import TestClient
|
|
from psycopg2.pool import ThreadedConnectionPool
|
|
import psycopg2
|
|
|
|
from app.core.config import settings
|
|
from app.core import database
|
|
from app.core.auth_dependencies import get_current_user
|
|
from app.core.auth_service import AuthService
|
|
from app.modules.remote import security, service
|
|
from app.modules.remote.router import router, views, Command
|
|
|
|
|
|
def test_code_preserves_leading_zeroes(monkeypatch):
|
|
monkeypatch.setattr(security.secrets, 'randbelow', lambda maximum: 42)
|
|
assert security.new_code() == '000000000042'
|
|
|
|
|
|
def test_day_is_copenhagen_calendar_day():
|
|
assert str(security.today(datetime(2026, 10, 13, 22, 30, tzinfo=timezone.utc))) == '2026-10-14'
|
|
assert str(security.today(datetime(2026, 1, 13, 22, 30, tzinfo=timezone.utc))) == '2026-01-13'
|
|
|
|
|
|
@pytest.mark.parametrize('seconds,status', [(3601, 'normal'), (3600, 'warning'), (0, 'warning'), (-1, 'overdue')])
|
|
def test_deadline_thresholds(seconds, status):
|
|
now = datetime(2026, 10, 13, 12, tzinfo=timezone.utc)
|
|
row = service.case_display({'deadline': now + timedelta(seconds=seconds)}, now)
|
|
assert row['deadline_state'] == status
|
|
|
|
|
|
def test_naive_deadline_uses_danish_timezone():
|
|
row = service.case_display({'deadline': datetime(2026, 10, 13, 14)}, datetime(2026, 10, 13, 12, tzinfo=timezone.utc))
|
|
assert row['deadline_seconds'] == 0
|
|
assert service.case_display({'deadline': None})['deadline_state'] == 'none'
|
|
|
|
|
|
@pytest.mark.parametrize("scheme", ["http", "https"])
|
|
def test_flag_origin_and_csrf(monkeypatch, scheme):
|
|
app = FastAPI(); app.include_router(router)
|
|
client = TestClient(app, base_url=f'{scheme}://hub.test')
|
|
monkeypatch.setattr(settings, 'REMOTE_ENABLED', False)
|
|
assert client.get('/api/v1/remote/bootstrap').status_code == 404
|
|
monkeypatch.setattr(settings, 'REMOTE_ENABLED', True)
|
|
monkeypatch.setattr(settings, 'REMOTE_LOOKUP_KEY', '')
|
|
assert client.get('/api/v1/remote/bootstrap').status_code == 503
|
|
monkeypatch.setattr(settings, 'REMOTE_LOOKUP_KEY', 'test-key-only-' * 4)
|
|
response = client.get('/api/v1/remote/bootstrap')
|
|
assert response.status_code == 200
|
|
assert ('Secure' in response.headers['set-cookie']) == (scheme == 'https')
|
|
assert 'HttpOnly' in response.headers['set-cookie']
|
|
assert client.post('/api/v1/remote/connect', json={'code': '0'*12, 'pin': '1234'}).status_code == 403
|
|
assert client.post('/api/v1/remote/connect', headers={'Origin': 'https://other.test'}, json={'pin': '1234'}).status_code == 403
|
|
|
|
|
|
def test_remote_token_is_not_hub_jwt():
|
|
assert AuthService.verify_token('remote-device-random-token') is None
|
|
|
|
|
|
@pytest.fixture
|
|
def db(monkeypatch):
|
|
import os
|
|
if os.environ.get('BMC_REMOTE_DB_TESTS') != '1':
|
|
pytest.skip('Set BMC_REMOTE_DB_TESTS=1 for isolated local PostgreSQL integration tests')
|
|
assert urlsplit(settings.DATABASE_URL).hostname in {'localhost', '127.0.0.1', '::1'}
|
|
schema = 'remote_test_' + uuid4().hex
|
|
root = psycopg2.connect(settings.DATABASE_URL)
|
|
root.autocommit = True
|
|
with root.cursor() as cur:
|
|
cur.execute('CREATE SCHEMA ' + schema)
|
|
pool = ThreadedConnectionPool(1, 12, dsn=settings.DATABASE_URL, options='-c search_path=' + schema)
|
|
monkeypatch.setattr(database, 'connection_pool', pool)
|
|
monkeypatch.setattr(settings, 'REMOTE_ENABLED', True)
|
|
monkeypatch.setattr(settings, 'REMOTE_LOOKUP_KEY', 'isolated-test-key-only-' * 3)
|
|
monkeypatch.setattr(settings, 'WORKFLOWS_ENABLED', False)
|
|
permissions = {1: ['cases.view'], 2: ['cases.view']}
|
|
monkeypatch.setattr(AuthService, 'get_user_permissions', lambda uid: permissions.get(uid, []))
|
|
monkeypatch.setattr(service, 'get_user_messages_summary', lambda uid: {'count': 0, 'list': []})
|
|
try:
|
|
with database.atomic_transaction():
|
|
database.execute_query('''CREATE TABLE users(user_id INTEGER PRIMARY KEY,username TEXT,full_name TEXT,email TEXT,is_2fa_enabled BOOLEAN DEFAULT FALSE,is_superadmin BOOLEAN DEFAULT FALSE,is_active BOOLEAN DEFAULT TRUE);
|
|
CREATE TABLE sessions(user_id INTEGER,token_jti TEXT,expires_at TIMESTAMP,revoked BOOLEAN DEFAULT FALSE);
|
|
CREATE TABLE customers(id INTEGER PRIMARY KEY,name TEXT);
|
|
CREATE TABLE sag_sager(id INTEGER PRIMARY KEY,titel TEXT,status TEXT,deadline TIMESTAMP,customer_id INTEGER,ansvarlig_bruger_id INTEGER,deleted_at TIMESTAMP,created_at TIMESTAMP DEFAULT NOW());
|
|
CREATE TABLE contacts(id INTEGER PRIMARY KEY,first_name TEXT,last_name TEXT,email TEXT,phone TEXT,mobile TEXT,title TEXT,department TEXT,user_company TEXT,is_active BOOLEAN,created_at TIMESTAMP,updated_at TIMESTAMP);
|
|
CREATE TABLE contact_companies(contact_id INTEGER,customer_id INTEGER);
|
|
INSERT INTO contacts(id,first_name,last_name,phone,mobile,is_active) VALUES(1,'Test','Kontakt','+45 12345678','87654321',TRUE),(2,'Inaktiv','Kontakt','12345678',NULL,FALSE);
|
|
CREATE TABLE settings(key TEXT,value TEXT);
|
|
CREATE TABLE tmodule_times (LIKE public.tmodule_times INCLUDING DEFAULTS);
|
|
ALTER TABLE tmodule_times ALTER COLUMN id DROP DEFAULT;
|
|
ALTER TABLE tmodule_times ALTER COLUMN id ADD GENERATED BY DEFAULT AS IDENTITY;
|
|
CREATE FUNCTION remote_test_update() RETURNS trigger LANGUAGE plpgsql AS $$BEGIN NEW.updated_at=clock_timestamp();RETURN NEW;END$$;
|
|
CREATE TRIGGER remote_test_update BEFORE UPDATE ON tmodule_times FOR EACH ROW EXECUTE FUNCTION remote_test_update();
|
|
INSERT INTO users(user_id,username,full_name) VALUES(1,'test1','Test One'),(2,'test2','Test Two');
|
|
INSERT INTO sag_sager(id,titel,status,ansvarlig_bruger_id) VALUES(11,'Ny testsag','Ny',1),(12,'Anden sag','I gang',1),(13,'Lukket sag','Lukket',1),(14,'Andens sag','Ny',2);''')
|
|
database.execute_query(Path('migrations/1069_remote_control.sql').read_text())
|
|
yield permissions
|
|
finally:
|
|
pool.closeall()
|
|
with root.cursor() as cur:
|
|
cur.execute('DROP SCHEMA ' + schema + ' CASCADE')
|
|
root.close()
|
|
|
|
|
|
@pytest.fixture(params=["http", "https"])
|
|
def web(db, request):
|
|
app = FastAPI(); app.include_router(router); app.include_router(views)
|
|
async def hub_user(request: Request):
|
|
uid = request.cookies.get('access_token')
|
|
if uid not in {'hub1', 'hub2'}:
|
|
raise HTTPException(401, 'Hub login required')
|
|
uid = int(uid[-1])
|
|
return {'id': uid, 'username': 'test' + str(uid), 'full_name': 'Test', 'permissions': db[uid]}
|
|
app.dependency_overrides[get_current_user] = hub_user
|
|
@app.get('/ordinary')
|
|
async def ordinary(user=__import__('fastapi').Depends(get_current_user)):
|
|
return {'id': user['id']}
|
|
client = TestClient(app, base_url=f'{request.param}://hub.test')
|
|
client.cookies.set('access_token', 'hub1')
|
|
csrf = client.get('/api/v1/remote/bootstrap').json()['csrf']
|
|
headers = {'Origin': str(client.base_url).rstrip('/'), 'X-Remote-CSRF': csrf}
|
|
return client, headers
|
|
|
|
|
|
def connect(web):
|
|
client, headers = web
|
|
assert client.put('/api/v1/remote/admin/pin', json={'pin': '1234'}, headers=headers).status_code == 200
|
|
remote = client.post('/api/v1/remote/admin/remotes', json={'name': 'Test remote'}, headers=headers).json()
|
|
response = client.post('/api/v1/remote/connect', json={'code': remote['code'], 'pin': '1234'}, headers=headers)
|
|
assert response.status_code == 200, response.text
|
|
assert ('Secure' in response.headers['set-cookie']) == (client.base_url.scheme == 'https')
|
|
assert 'HttpOnly' in response.headers['set-cookie']
|
|
return remote
|
|
|
|
|
|
def post_command(web, action, **kwargs):
|
|
client, headers = web
|
|
snapshot = client.get('/api/v1/remote/state').json()
|
|
payload = {'request_id': str(uuid4()), 'version': snapshot['version'], 'timer_version': snapshot['timer_version'], 'action': action, **kwargs}
|
|
return client.post('/api/v1/remote/command', json=payload, headers=headers), payload
|
|
|
|
|
|
def test_connection_isolation_pin_rotation_expiry(web, db):
|
|
client, headers = web; remote = connect(web)
|
|
assert client.get('/api/v1/remote/state').status_code == 200
|
|
# A remote cookie alone grants no full Hub or management access.
|
|
client.cookies.delete('access_token')
|
|
assert client.get('/ordinary').status_code == 401
|
|
assert client.get('/api/v1/remote/admin').status_code == 401
|
|
client.cookies.set('access_token', 'hub2')
|
|
assert client.post(f"/api/v1/remote/admin/remotes/{remote['id']}/rotate", json={}, headers=headers).status_code == 404
|
|
assert client.get('/api/v1/remote/admin').json()['remotes'] == []
|
|
client.cookies.set('access_token', 'hub1')
|
|
assert client.put('/api/v1/remote/admin/pin', json={'pin': '5678'}, headers=headers).status_code == 200
|
|
assert client.get('/api/v1/remote/state').status_code == 423
|
|
database.execute_query("UPDATE remote_attempts SET next_allowed_at='epoch'")
|
|
assert client.post('/api/v1/remote/unlock', json={'pin': '5678'}, headers=headers).status_code == 200
|
|
database.execute_query("UPDATE remote_devices SET unlocked_date=unlocked_date-1")
|
|
assert client.get('/api/v1/remote/state').status_code == 423
|
|
database.execute_query("UPDATE remote_devices SET expires_at=NOW()-INTERVAL '1 second'")
|
|
assert client.get('/api/v1/remote/state').status_code == 401
|
|
|
|
|
|
def test_failed_attempts_are_committed_and_shared_between_connections(web):
|
|
client, headers = web
|
|
assert client.post('/api/v1/remote/connect', json={'code': '000000000042', 'pin': '0000'}, headers=headers).status_code == 401
|
|
second = TestClient(client.app, base_url=str(client.base_url)); second.cookies.update(client.cookies)
|
|
result = second.post('/api/v1/remote/connect', json={'code': '000000000042', 'pin': '0000'}, headers=headers)
|
|
assert result.status_code == 429 and 1 <= int(result.headers['Retry-After']) <= 30
|
|
assert database.execute_query_single('SELECT COUNT(*) AS count FROM remote_attempts')['count'] == 1
|
|
|
|
|
|
def test_rate_limit_is_atomic_across_concurrent_db_connections(db):
|
|
def attempt():
|
|
try:
|
|
security.reserve_attempt('isolated-ip', 1, uuid4())
|
|
return 200
|
|
except HTTPException as exc:
|
|
return exc.status_code
|
|
with ThreadPoolExecutor(max_workers=4) as executor:
|
|
outcomes = list(executor.map(lambda _: attempt(), range(4)))
|
|
assert sorted(outcomes) == [200, 429, 429, 429]
|
|
|
|
|
|
def process_attempt(options):
|
|
"""Separate spawned worker: no shared Python lock, pool, or rate-limit memory."""
|
|
settings.REMOTE_LOOKUP_KEY = 'isolated-test-key-only-' * 3
|
|
database.connection_pool = ThreadedConnectionPool(1, 2, dsn=settings.DATABASE_URL, options=options)
|
|
try:
|
|
try:
|
|
security.reserve_attempt('cross-process-ip', 1)
|
|
return 200
|
|
except HTTPException as exc:
|
|
return exc.status_code
|
|
finally:
|
|
database.connection_pool.closeall()
|
|
|
|
|
|
def test_rate_limit_is_shared_between_processes(db):
|
|
import multiprocessing
|
|
conn = database.get_db_connection()
|
|
options = conn.get_dsn_parameters()['options']
|
|
database.release_db_connection(conn)
|
|
with ProcessPoolExecutor(max_workers=2, mp_context=multiprocessing.get_context('spawn')) as executor:
|
|
assert sorted(executor.map(process_attempt, [options, options])) == [200, 429]
|
|
|
|
|
|
def test_timer_actions_manual_time_idempotency_and_conflict(web):
|
|
client, headers = web; connect(web)
|
|
result, payload = post_command(web, 'start', case_id=11)
|
|
assert result.status_code == 200, result.text
|
|
assert client.post('/api/v1/remote/command', json=payload, headers=headers).json() == result.json()
|
|
assert client.get('/api/v1/remote/command-result/' + payload['request_id']).json()['completed'] is True
|
|
assert database.execute_query_single('SELECT COUNT(*) AS count FROM tmodule_times')['count'] == 1
|
|
payload['case_id'] = 12
|
|
assert client.post('/api/v1/remote/command', json=payload, headers=headers).status_code == 409
|
|
assert post_command(web, 'select', case_id=12)[0].status_code == 409
|
|
assert post_command(web, 'select', case_id=12, switch_timer=False)[0].status_code == 200
|
|
assert client.get('/api/v1/remote/state').json()['timer']['active']['sag_id'] == 11
|
|
assert post_command(web, 'select', case_id=12, switch_timer=True)[0].status_code == 200
|
|
snapshot = client.get('/api/v1/remote/state').json()
|
|
assert snapshot['timer']['active']['sag_id'] == 12
|
|
assert post_command(web, 'pause')[0].status_code == 200
|
|
tid = snapshot['timer']['active']['time_entry_id']
|
|
assert post_command(web, 'resume', time_id=tid)[0].status_code == 200
|
|
assert post_command(web, 'stop', time_id=tid, description='Test arbejde')[0].status_code == 200
|
|
assert post_command(web, 'manual', case_id=12, description='Test manuel tid', minutes=17)[0].status_code == 200
|
|
assert client.get('/api/v1/remote/state').json()['today_minutes'] >= 17
|
|
# Desktop modifications invalidate an already prepared remote command.
|
|
snapshot = client.get('/api/v1/remote/state').json()
|
|
payload = dict(request_id=str(uuid4()), version=snapshot['version'], timer_version=snapshot['timer_version'], action='start', case_id=11)
|
|
import importlib
|
|
timers = importlib.import_module('app.timetracking.backend.router')
|
|
asyncio.run(timers.start_live_timer_v1({'sag_id': 12, 'billing_mode': 'automatic'}, {'id': 1, 'username': 'test1'}))
|
|
assert client.post('/api/v1/remote/command', json=payload, headers=headers).status_code == 409
|
|
|
|
|
|
def test_concurrent_commands_cannot_create_duplicate_timers(web):
|
|
client, headers = web; connect(web)
|
|
snapshot = client.get('/api/v1/remote/state').json()
|
|
def start(_):
|
|
payload = Command(request_id=uuid4(), version=snapshot['version'], timer_version=snapshot['timer_version'], action='start', case_id=11)
|
|
try:
|
|
asyncio.run(service.command(payload, {'id': 1, 'username': 'test1', 'permissions': ['cases.view']}))
|
|
return 200
|
|
except HTTPException as exc:
|
|
return exc.status_code
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
assert sorted(executor.map(start, range(2))) == [200, 409]
|
|
assert database.execute_query_single('SELECT COUNT(*) AS count FROM tmodule_times WHERE aktiv_timer')['count'] == 1
|
|
|
|
|
|
def test_only_selected_follower_and_no_stale_selection(web):
|
|
client, headers = web; connect(web)
|
|
first, second = str(uuid4()), str(uuid4())
|
|
client.post('/api/v1/remote/follow', json={'tab_id': first, 'enable': True}, headers=headers)
|
|
assert post_command(web, 'select', case_id=11)[0].status_code == 200
|
|
event = client.post('/api/v1/remote/follow', json={'tab_id': first}, headers=headers).json()
|
|
assert event['selection'] == {'case_id': 11}
|
|
client.post('/api/v1/remote/follow', json={'tab_id': second, 'enable': True}, headers=headers)
|
|
assert client.post('/api/v1/remote/follow', json={'tab_id': first}, headers=headers).json()['following'] is False
|
|
assert post_command(web, 'select', case_id=12)[0].status_code == 200
|
|
database.execute_query("UPDATE remote_user_state SET follower_seen_at=NOW()-INTERVAL '15 seconds'")
|
|
event = client.post('/api/v1/remote/follow', json={'tab_id': second}, headers=headers).json()
|
|
assert event['selection'] is None
|
|
|
|
|
|
def test_case_filters_permissions_and_revocation(web, db):
|
|
client, headers = web; remote = connect(web)
|
|
assert {c['id'] for c in client.get('/api/v1/remote/cases?kind=new').json()['items']} == {11,14}
|
|
assert {c['id'] for c in client.get('/api/v1/remote/cases?kind=own').json()['items']} == {11,12}
|
|
assert client.get('/api/v1/remote/cases?q=11').json()['items'][0]['id'] == 11
|
|
db[1] = []
|
|
assert client.get('/api/v1/remote/cases').status_code == 403
|
|
assert post_command(web, 'start', case_id=11)[0].status_code == 403
|
|
db[1] = ['cases.view']
|
|
assert client.request('DELETE', f"/api/v1/remote/admin/remotes/{remote['id']}", json={}, headers=headers).status_code == 200
|
|
assert client.get('/api/v1/remote/state').status_code == 401
|
|
|
|
|
|
def test_leading_zero_code_and_no_hashes_in_administration(web, monkeypatch):
|
|
client, headers = web
|
|
monkeypatch.setattr('app.modules.remote.router.new_code', lambda: '000000000042')
|
|
remote = connect(web)
|
|
assert remote['code'] == '000000000042'
|
|
admin = client.get('/api/v1/remote/admin').json()
|
|
assert not {'code', 'code_hash', 'code_lookup', 'pin_hash', 'token_hash'}.intersection(admin['remotes'][0])
|
|
assert not {'token_hash', 'pin_hash'}.intersection(admin['devices'][0])
|
|
device_id = admin['devices'][0]['id']
|
|
assert client.request('DELETE', '/api/v1/remote/admin/devices/' + device_id, json={}, headers=headers).status_code == 200
|
|
assert client.get('/api/v1/remote/state').status_code == 401
|
|
|
|
|
|
def test_rotate_code_revokes_devices_and_inactive_user_is_denied(web):
|
|
client, headers = web; remote = connect(web)
|
|
result = client.post(f"/api/v1/remote/admin/remotes/{remote['id']}/rotate", json={}, headers=headers)
|
|
assert result.status_code == 200 and result.json()['code'] != remote['code']
|
|
assert client.get('/api/v1/remote/state').status_code == 401
|
|
database.execute_query("UPDATE remote_attempts SET next_allowed_at='epoch'")
|
|
assert client.post('/api/v1/remote/connect', json={'code': result.json()['code'], 'pin': '1234'}, headers=headers).status_code == 200
|
|
database.execute_query('UPDATE users SET is_active=FALSE WHERE user_id=1')
|
|
assert client.get('/api/v1/remote/state').status_code == 401
|
|
|
|
|
|
def test_wrong_pin_and_reconnect_do_not_bypass_cooldown(web):
|
|
client, headers = web; connect(web)
|
|
database.execute_query("UPDATE remote_attempts SET next_allowed_at='epoch'")
|
|
database.execute_query('UPDATE remote_devices SET unlocked_date=unlocked_date-1')
|
|
assert client.post('/api/v1/remote/unlock', json={'pin': '9999'}, headers=headers).status_code == 401
|
|
assert client.post('/api/v1/remote/unlock', json={'pin': '1234'}, headers=headers).status_code == 429
|
|
assert client.get('/api/v1/remote/state').status_code == 423
|
|
|
|
|
|
def test_message_read_is_explicit_and_uses_current_user(web, monkeypatch):
|
|
client, _ = web; connect(web)
|
|
calls = []
|
|
monkeypatch.setattr(service, 'mark_user_messages_read', lambda uid, partner_user_id=None: calls.append((uid,partner_user_id)) or 1)
|
|
assert client.get('/api/v1/remote/state').status_code == 200
|
|
assert calls == []
|
|
assert post_command(web, 'read', partner_user_id=2)[0].status_code == 200
|
|
assert calls == [(1,2)]
|
|
|
|
|
|
def test_desktop_start_and_remote_start_share_user_lock(web):
|
|
client, _ = web; connect(web)
|
|
snapshot = client.get('/api/v1/remote/state').json()
|
|
user = {'id': 1, 'username': 'test1', 'permissions': ['cases.view']}
|
|
def start_remote():
|
|
payload = Command(request_id=uuid4(), version=snapshot['version'], timer_version=snapshot['timer_version'], action='start', case_id=11)
|
|
try:
|
|
asyncio.run(service.command(payload,user))
|
|
return 200
|
|
except HTTPException as exc:
|
|
return exc.status_code
|
|
def start_desktop():
|
|
import importlib
|
|
timers = importlib.import_module('app.timetracking.backend.router')
|
|
asyncio.run(timers.start_live_timer_v1({'sag_id':12,'billing_mode':'automatic'},user))
|
|
return 200
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
outcomes = [executor.submit(start_remote),executor.submit(start_desktop)]
|
|
assert all(f.result() in {200,409} for f in outcomes)
|
|
assert database.execute_query_single('SELECT COUNT(*) AS count FROM tmodule_times WHERE aktiv_timer')['count'] == 1
|
|
|
|
|
|
def test_real_hub_jwt_remains_separate_and_revocable(web):
|
|
client, _ = web; connect(web)
|
|
client.app.dependency_overrides.clear()
|
|
token = AuthService.create_access_token(1, 'test1')
|
|
client.cookies.set('access_token', token)
|
|
assert client.get('/ordinary').json() == {'id': 1}
|
|
assert client.get('/api/v1/remote/admin').status_code == 200
|
|
client.cookies.set('access_token', client.cookies.get(security.COOKIE))
|
|
assert client.get('/ordinary').status_code == 401
|
|
client.cookies.set('access_token', token)
|
|
database.execute_query('UPDATE sessions SET revoked=TRUE')
|
|
assert client.get('/ordinary').status_code == 401
|
|
assert client.get('/api/v1/remote/admin').status_code == 401
|
|
assert client.get('/api/v1/remote/state').status_code == 200
|
|
|
|
|
|
def test_remote_contacts_and_owner_only_call(web, db, monkeypatch):
|
|
from app.modules.telefoni.backend import router as phone_router
|
|
client, headers = web; connect(web)
|
|
assert client.get('/api/v1/remote/contacts').status_code == 403
|
|
db[1].append('customers.view')
|
|
result = client.get('/api/v1/remote/contacts?q=Kontakt').json()
|
|
assert len(result['items']) == 1 and result['items'][0]['id'] == 1
|
|
calls = []
|
|
async def fake_call(payload, request):
|
|
calls.append((payload.model_dump(), request.state.user_id))
|
|
return {'action_url': 'secret-must-not-be-returned'}
|
|
monkeypatch.setattr(phone_router, 'click_to_call', fake_call)
|
|
payload = {'request_id': str(uuid4()), 'contact_id': 1, 'field': 'mobile'}
|
|
first = client.post('/api/v1/remote/call', json=payload, headers=headers)
|
|
assert first.status_code == 200 and first.json()['status'] == 'ok'
|
|
assert 'secret' not in first.text
|
|
assert client.post('/api/v1/remote/call', json=payload, headers=headers).json() == first.json()
|
|
assert len(calls) == 1 and calls[0][0]['user_id'] == 1 and calls[0][1] == 1
|
|
assert calls[0][0]['number'] == '87654321' and calls[0][0]['extension'] is None
|
|
assert client.get('/api/v1/remote/call/' + payload['request_id']).json() == first.json()
|
|
assert client.post('/api/v1/remote/call', json={**payload, 'user_id': 2}, headers=headers).status_code == 422
|
|
assert client.post('/api/v1/remote/call', json={**payload, 'field': 'phone'}, headers=headers).status_code == 409
|
|
assert client.post('/api/v1/remote/call', json={**payload, 'request_id': str(uuid4()), 'contact_id': 2}, headers=headers).status_code == 404
|
|
|
|
|
|
def test_uncertain_call_is_not_repeated(web, db, monkeypatch):
|
|
from app.modules.telefoni.backend import router as phone_router
|
|
client, headers = web; connect(web); db[1].append('customers.view')
|
|
calls = []
|
|
async def failed_call(payload, request):
|
|
calls.append(payload)
|
|
raise HTTPException(502, 'timeout')
|
|
monkeypatch.setattr(phone_router, 'click_to_call', failed_call)
|
|
payload = {'request_id': str(uuid4()), 'contact_id': 1, 'field': 'phone'}
|
|
assert client.post('/api/v1/remote/call', json=payload, headers=headers).json()['status'] == 'uncertain'
|
|
assert client.post('/api/v1/remote/call', json=payload, headers=headers).json()['status'] == 'uncertain'
|
|
assert len(calls) == 1
|
|
assert client.get('/api/v1/remote/call/' + str(uuid4())).json()['status'] == 'unknown'
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_incoming_calls_use_shared_state_and_ignore_stale(monkeypatch):
|
|
from app.modules.remote import telephony
|
|
now = datetime.now(timezone.utc)
|
|
monkeypatch.setattr(telephony.MissionService, 'get_active_calls', lambda: [
|
|
{'call_id':'fresh','updated_at':now,'caller_number':'12345678','ringing_phones':['Support']},
|
|
{'call_id':'stale','updated_at':now-timedelta(minutes=10)}])
|
|
user = {'id': 1, 'permissions': ['customers.view']}
|
|
assert [c['call_id'] for c in telephony.status(user)['incoming']] == ['fresh']
|
|
assert telephony.status({'id':2,'permissions':[]}) == {'allowed':False,'incoming':[]}
|
|
|
|
|
|
def test_concurrent_same_call_reserves_before_external_action(web, db, monkeypatch):
|
|
from app.modules.telefoni.backend import router as phone_router
|
|
client, headers = web; connect(web); db[1].append('customers.view')
|
|
calls = []
|
|
async def slow_call(payload, request):
|
|
calls.append(payload.user_id)
|
|
await asyncio.sleep(0.1)
|
|
return {'status': 'ok'}
|
|
monkeypatch.setattr(phone_router, 'click_to_call', slow_call)
|
|
payload = {'request_id': str(uuid4()), 'contact_id': 1, 'field': 'phone'}
|
|
def send(_):
|
|
with TestClient(client.app, base_url=str(client.base_url)) as other:
|
|
other.cookies.update(client.cookies)
|
|
return other.post('/api/v1/remote/call', json=payload, headers=headers)
|
|
with ThreadPoolExecutor(max_workers=2) as workers:
|
|
responses = list(workers.map(send, range(2)))
|
|
assert all(response.status_code == 200 for response in responses)
|
|
assert calls == [1]
|
|
assert client.get('/api/v1/remote/call/' + payload['request_id']).json()['status'] == 'ok'
|
|
|
|
|
|
def test_remote_planned_day_is_scoped_to_user_and_danish_day(monkeypatch):
|
|
from datetime import date
|
|
captured = []
|
|
monkeypatch.setattr(service, 'case_permission', lambda user: None)
|
|
monkeypatch.setattr(service, 'execute_query', lambda sql, params: captured.append((sql, params)) or [])
|
|
result = service.planned_day({'id': 7}, date(2026, 10, 14))
|
|
assert result == {'date': '2026-10-14', 'items': []}
|
|
sql, params = captured[0]
|
|
assert params[0] == params[3] == 7
|
|
assert params[1].hour == params[2].hour == 0
|
|
assert params[1].day == 15 and params[2].day == 14
|
|
assert params[2].utcoffset() == timedelta(hours=2)
|
|
assert "q.state='planned'" in sql and 'a.deleted_at IS NULL' in sql
|
|
|
|
|
|
def test_remote_planned_day_uses_dst_boundaries(monkeypatch):
|
|
from datetime import date
|
|
calls = []
|
|
monkeypatch.setattr(service, 'case_permission', lambda user: None)
|
|
monkeypatch.setattr(service, 'execute_query', lambda sql, params: calls.append(params) or [])
|
|
service.planned_day({'id': 7}, date(2026, 10, 25))
|
|
end, start = calls[0][1:3]
|
|
assert (end.astimezone(timezone.utc) - start.astimezone(timezone.utc)) == timedelta(hours=25)
|