147 lines
9.1 KiB
Python
147 lines
9.1 KiB
Python
"""Machine linking tests. Database verification is opt-in and always rolls back."""
|
|
import asyncio
|
|
from contextlib import contextmanager
|
|
import json
|
|
import os
|
|
import time
|
|
import unittest
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
from uuid import uuid4
|
|
|
|
from fastapi import HTTPException
|
|
from app.services import anydesk_links as links
|
|
|
|
|
|
class MachineIdentityTests(unittest.TestCase):
|
|
def test_only_customer_end_is_used(self):
|
|
self.assertEqual(links.customer_machine_id({'to_id': '123 456-789', 'from_id': '999'}), '123456789')
|
|
self.assertIsNone(links.customer_machine_id({'from_id': '123456789', 'remote_id': '123456789'}))
|
|
self.assertIsNone(links.customer_machine_id({'to_id': 'computer123'}))
|
|
self.assertEqual(links.customer_machine_id('{"to_id":"123456789"}'), '123456789')
|
|
|
|
def test_missing_customer_id_cannot_create_mapping(self):
|
|
with patch.object(links, 'execute_query', side_effect=[
|
|
[{'device_info': {'to_id': '123456789'}}], []
|
|
]):
|
|
with self.assertRaises(HTTPException) as error:
|
|
links.remember_machine(1, None, None)
|
|
self.assertEqual(error.exception.status_code, 409)
|
|
|
|
def test_missing_customer_machine_id_does_not_use_technician(self):
|
|
with patch.object(links, 'execute_query', return_value=[{'device_info': {'from_id': '999'}}]):
|
|
with self.assertRaises(HTTPException):
|
|
links.remember_machine(1, 2, 3)
|
|
|
|
def test_local_bridge_applies_mapping_but_never_during_dry_run(self):
|
|
from app.services import anydesk as service
|
|
for dry_run in (False, True):
|
|
with self.subTest(dry_run=dry_run):
|
|
client = AsyncMock()
|
|
client.__aenter__.return_value = client
|
|
client.get.return_value = Mock(json=Mock(return_value=[]))
|
|
with patch.object(service.httpx, 'AsyncClient', return_value=client), \
|
|
patch.object(service, 'apply_machine_links') as apply:
|
|
result = asyncio.run(service.AnyDeskService().fetch_sessions_from_local_endpoint(
|
|
'http://test.invalid/sessions', dry_run=dry_run))
|
|
self.assertEqual(result['errors'], [])
|
|
if dry_run:
|
|
apply.assert_not_called()
|
|
else:
|
|
apply.assert_called_once_with()
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get('BMC_ANYDESK_DB_TESTS') == '1', 'Opt-in rollback-only database test')
|
|
class MachineDatabaseTests(unittest.TestCase):
|
|
def test_manual_registration_backfills_and_next_import_remembers(self):
|
|
import psycopg2
|
|
from psycopg2.extras import RealDictCursor, Json
|
|
from app.core.config import settings
|
|
from app.routers import anydesk as router
|
|
from app.services import anydesk as service
|
|
conn = psycopg2.connect(settings.DATABASE_URL)
|
|
def query(sql, params=None):
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
|
|
cursor.execute(sql, params)
|
|
return list(cursor.fetchall()) if cursor.description else cursor.rowcount
|
|
@contextmanager
|
|
def transaction():
|
|
yield conn
|
|
token = uuid4().hex
|
|
machine = str(uuid4().int)[:24]
|
|
tech = str(uuid4().int)[:24]
|
|
try:
|
|
pairs = query('''SELECT cc.customer_id,cc.contact_id FROM contact_companies cc
|
|
JOIN customers c ON c.id=cc.customer_id AND c.is_active=TRUE
|
|
JOIN contacts p ON p.id=cc.contact_id AND p.is_active=TRUE LIMIT 1''')
|
|
self.assertTrue(pairs, 'Requires an existing active customer/contact pair')
|
|
customer = pairs[0]['customer_id']
|
|
contact = query("INSERT INTO contacts(first_name,last_name,is_active) VALUES('AnyDesk test',%s,TRUE) RETURNING id", (token,))[0]['id']
|
|
query('INSERT INTO contact_companies(contact_id,customer_id) VALUES(%s,%s)', (contact, customer))
|
|
other = query('SELECT id FROM customers WHERE is_active=TRUE AND id<>%s LIMIT 1', (customer,))[0]['id']
|
|
def session(label, to_id=machine, cust=None, person=None):
|
|
return query('''INSERT INTO anydesk_sessions
|
|
(anydesk_session_id,device_info,customer_id,contact_id,status,notes)
|
|
VALUES(%s,%s,%s,%s,'completed','keep this note') RETURNING id''',
|
|
(token+label, Json({'to_id': to_id, 'from_id': tech}), cust, person))[0]['id']
|
|
selected = session('selected')
|
|
previous = session('previous')
|
|
conflict = session('conflict', cust=other)
|
|
technician = session('technician', to_id=tech)
|
|
no_customer_end = session('missing', to_id=None)
|
|
with patch.object(links, 'execute_query', query), patch.object(router, 'execute_query', query), \
|
|
patch.object(router, 'atomic_transaction', transaction), patch.object(service, 'execute_query', query):
|
|
response = asyncio.run(router.link_session(selected, customer_id=customer, contact_id=contact,
|
|
remember_machine_link=True))
|
|
result = json.loads(response.body)
|
|
self.assertTrue(result['remembered'])
|
|
self.assertEqual(result['linked_sessions'], 1)
|
|
hardware = query('SELECT id,asset_type,model,current_owner_customer_id FROM hardware_assets WHERE anydesk_id=%s', (machine,))
|
|
self.assertEqual(len(hardware), 1)
|
|
self.assertEqual((hardware[0]['asset_type'], hardware[0]['model']), ('pc', 'PC'))
|
|
hardware_id = hardware[0]['id']
|
|
self.assertTrue(query('SELECT 1 FROM hardware_contacts WHERE hardware_id=%s AND contact_id=%s', (hardware_id, contact)))
|
|
asyncio.run(router.link_session(selected, customer_id=customer, contact_id=contact, remember_machine_link=True))
|
|
self.assertEqual(len(query('SELECT id FROM hardware_assets WHERE anydesk_id=%s', (machine,))), 1)
|
|
saved = query('SELECT customer_id,contact_id,notes,sag_id FROM anydesk_sessions WHERE id=%s', (previous,))[0]
|
|
self.assertEqual((saved['customer_id'], saved['contact_id']), (customer, contact))
|
|
self.assertEqual(saved['notes'], 'keep this note')
|
|
self.assertIsNone(saved['sag_id'])
|
|
self.assertEqual(query('SELECT customer_id FROM anydesk_sessions WHERE id=%s', (conflict,))[0]['customer_id'], other)
|
|
for row_id in (technician, no_customer_end):
|
|
self.assertIsNone(query('SELECT customer_id FROM anydesk_sessions WHERE id=%s', (row_id,))[0]['customer_id'])
|
|
api = service.AnyDeskService()
|
|
api._api_call = AsyncMock(return_value={'list': [
|
|
{'sid': token+'future', 'from': {'cid': tech}, 'to': {'cid': machine},
|
|
'start-time': int(time.time()), 'duration': 120}
|
|
]})
|
|
imported = asyncio.run(api.fetch_sessions_from_api())
|
|
self.assertEqual(imported['errors'], [])
|
|
row = query('SELECT customer_id,contact_id FROM anydesk_sessions WHERE anydesk_session_id=%s', (token+'future',))[0]
|
|
self.assertEqual((row['customer_id'], row['contact_id']), (customer, contact))
|
|
self.assertEqual(links.apply_machine_links(machine), 0)
|
|
with self.assertRaises(HTTPException):
|
|
links.remember_machine(selected, other, contact)
|
|
# Reuse the contact's one unconfigured computer instead of duplicating it.
|
|
blank = query("INSERT INTO hardware_assets(asset_type,model,current_owner_type,current_owner_customer_id) VALUES('pc','Existing PC','customer',%s) RETURNING id", (customer,))[0]['id']
|
|
query("INSERT INTO hardware_contacts(hardware_id,contact_id) VALUES(%s,%s)", (blank, contact))
|
|
new_machine = str(uuid4().int)[:24]
|
|
new_session = session('blank', to_id=new_machine)
|
|
links.remember_machine(new_session, customer, contact)
|
|
self.assertEqual(query('SELECT anydesk_id,model FROM hardware_assets WHERE id=%s', (blank,))[0],
|
|
{'anydesk_id': new_machine, 'model': 'Existing PC'})
|
|
self.assertEqual(query('SELECT hardware_asset_id FROM anydesk_sessions WHERE id=%s', (new_session,))[0]['hardware_asset_id'], blank)
|
|
# Duplicate machine IDs must never be resolved by choosing an arbitrary PC.
|
|
query("INSERT INTO hardware_assets(asset_type,model,current_owner_customer_id,anydesk_id) VALUES('pc','Duplicate',%s,%s)", (customer, machine))
|
|
duplicate_session = session('duplicate')
|
|
self.assertEqual(links.apply_machine_links(machine), 0)
|
|
self.assertIsNone(query('SELECT hardware_asset_id FROM anydesk_sessions WHERE id=%s', (duplicate_session,))[0]['hardware_asset_id'])
|
|
with self.assertRaises(HTTPException):
|
|
links.remember_machine(duplicate_session, customer, contact)
|
|
finally:
|
|
conn.rollback()
|
|
conn.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|