63 lines
3.9 KiB
Python
63 lines
3.9 KiB
Python
|
import json
|
||
|
|
import os
|
||
|
|
import threading
|
||
|
|
import unittest
|
||
|
|
from http.client import HTTPConnection
|
||
|
|
from tempfile import TemporaryDirectory
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from app.main import create_server
|
||
|
|
|
||
|
|
|
||
|
|
class WebsiteScanApiTests(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.tmp = TemporaryDirectory()
|
||
|
|
os.environ['BOOTSTRAP_ADMIN_EMAIL'] = 'scan-owner@example.test'
|
||
|
|
os.environ['BOOTSTRAP_ADMIN_PASSWORD'] = 'password'
|
||
|
|
self.server = create_server('127.0.0.1', 0, self.tmp.name + '/db.sqlite')
|
||
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True); self.thread.start()
|
||
|
|
self.conn = HTTPConnection('127.0.0.1', self.server.server_port, timeout=4); self.cookie = None
|
||
|
|
self.request('POST', '/api/v1/auth/login', {'email': 'scan-owner@example.test', 'password': 'password'})
|
||
|
|
|
||
|
|
def tearDown(self):
|
||
|
|
self.server.shutdown(); self.server.server_close(); self.thread.join(2); self.tmp.cleanup()
|
||
|
|
|
||
|
|
def request(self, method, path, payload=None):
|
||
|
|
body = json.dumps(payload).encode() if payload is not None else None; headers = {'Content-Type': 'application/json'} if body else {}
|
||
|
|
if self.cookie: headers['Cookie'] = self.cookie
|
||
|
|
self.conn.request(method, path, body, headers); response = self.conn.getresponse(); cookie = response.getheader('Set-Cookie')
|
||
|
|
if cookie: self.cookie = cookie.split(';', 1)[0]
|
||
|
|
return response.status, json.loads(response.read() or b'{}')
|
||
|
|
|
||
|
|
def test_scan_is_cached_history_is_listed_and_audited(self):
|
||
|
|
_, business = self.request('POST', '/api/v1/businesses', {'name': 'Scan Co', 'website': 'https://scan.example'})
|
||
|
|
result = {'classification': 'healthy', 'status': 200, 'final_url': 'https://scan.example/', 'redirect_chain': []}
|
||
|
|
with patch('app.main.validate_url', return_value='https://scan.example/'), patch('app.main.scan_website', return_value=result) as scanner:
|
||
|
|
first_status, first = self.request('POST', f"/api/v1/businesses/{business['id']}/websites/scan", {})
|
||
|
|
second_status, second = self.request('POST', f"/api/v1/businesses/{business['id']}/websites/scan", {})
|
||
|
|
self.assertEqual(first_status, 201); self.assertEqual(second_status, 200); self.assertTrue(second['cache_hit']); scanner.assert_called_once()
|
||
|
|
status, history = self.request('GET', '/api/v1/website-scans?page_size=1')
|
||
|
|
self.assertEqual(status, 200); self.assertEqual(len(history['items']), 1); self.assertFalse(history['has_more'])
|
||
|
|
self.assertEqual(history['items'][0]['classification'], 'healthy')
|
||
|
|
self.assertEqual(self.request('GET', '/api/v1/website-scans?business_id=999999')[1]['items'], [])
|
||
|
|
|
||
|
|
def test_get_latest_scan_returns_scan_payload(self):
|
||
|
|
_, business = self.request('POST', '/api/v1/businesses', {'name': 'Read Scan Co', 'website': 'https://read.example'})
|
||
|
|
result = {'classification': 'unknown', 'status': 200, 'final_url': 'https://read.example/', 'redirect_chain': []}
|
||
|
|
with patch('app.main.validate_url', return_value='https://read.example/'), patch('app.main.scan_website', return_value=result):
|
||
|
|
self.assertEqual(self.request('POST', f"/api/v1/businesses/{business['id']}/websites/scan", {})[0], 201)
|
||
|
|
status, payload = self.request('GET', f"/api/v1/businesses/{business['id']}/websites/scan")
|
||
|
|
self.assertEqual(status, 200)
|
||
|
|
self.assertEqual(payload['classification'], 'unknown')
|
||
|
|
self.assertEqual(payload['business_id'], business['id'])
|
||
|
|
|
||
|
|
def test_unsafe_scan_is_rejected_without_fetching(self):
|
||
|
|
_, business = self.request('POST', '/api/v1/businesses', {'name': 'Private Scan', 'website': 'http://127.0.0.1/'})
|
||
|
|
with patch('app.main.scan_website') as scanner:
|
||
|
|
status, payload = self.request('POST', f"/api/v1/businesses/{business['id']}/websites/scan", {})
|
||
|
|
self.assertEqual(status, 400); self.assertEqual(payload['error'], 'unsafe_url'); scanner.assert_not_called()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
unittest.main()
|