sneakygeek/SneakyScan
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor to remove config_files in favor of db

Browse Source
This commit is contained in:
Phillip Tarrant
2025-11-19 20:29:14 -06:00
parent b2e6efb4b3
commit 41ba4c47b5
34 changed files with 463 additions and 536 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
app/tests/test_scan_service.py
View File

@@ -13,49 +13,42 @@ from web.services.scan_service import ScanService
class TestScanServiceTrigger:
"""Tests for triggering scans."""
def test_trigger_scan_valid_config(self, test_db, sample_config_file):
"""Test triggering a scan with valid config file."""
service = ScanService(test_db)
def test_trigger_scan_valid_config(self, db, sample_db_config):
"""Test triggering a scan with valid config."""
service = ScanService(db)
scan_id = service.trigger_scan(sample_config_file, triggered_by='manual')
scan_id = service.trigger_scan(config_id=sample_db_config.id, triggered_by='manual')
# Verify scan created
assert scan_id is not None
assert isinstance(scan_id, int)
# Verify scan in database
scan = test_db.query(Scan).filter(Scan.id == scan_id).first()
scan = db.query(Scan).filter(Scan.id == scan_id).first()
assert scan is not None
assert scan.status == 'running'
assert scan.title == 'Test Scan'
assert scan.triggered_by == 'manual'
assert scan.config_file == sample_config_file
assert scan.config_id == sample_db_config.id
def test_trigger_scan_invalid_config(self, test_db, sample_invalid_config_file):
"""Test triggering a scan with invalid config file."""
service = ScanService(test_db)
def test_trigger_scan_invalid_config(self, db):
"""Test triggering a scan with invalid config ID."""
service = ScanService(db)
with pytest.raises(ValueError, match="Invalid config file"):
service.trigger_scan(sample_invalid_config_file)
with pytest.raises(ValueError, match="not found"):
service.trigger_scan(config_id=99999)
def test_trigger_scan_nonexistent_file(self, test_db):
"""Test triggering a scan with nonexistent config file."""
service = ScanService(test_db)
with pytest.raises(ValueError, match="does not exist"):
service.trigger_scan('/nonexistent/config.yaml')
def test_trigger_scan_with_schedule(self, test_db, sample_config_file):
def test_trigger_scan_with_schedule(self, db, sample_db_config):
"""Test triggering a scan via schedule."""
service = ScanService(test_db)
service = ScanService(db)
scan_id = service.trigger_scan(
sample_config_file,
config_id=sample_db_config.id,
triggered_by='scheduled',
schedule_id=42
)
scan = test_db.query(Scan).filter(Scan.id == scan_id).first()
scan = db.query(Scan).filter(Scan.id == scan_id).first()
assert scan.triggered_by == 'scheduled'
assert scan.schedule_id == 42
@@ -63,19 +56,19 @@ class TestScanServiceTrigger:
class TestScanServiceGet:
"""Tests for retrieving scans."""
def test_get_scan_not_found(self, test_db):
def test_get_scan_not_found(self, db):
"""Test getting a nonexistent scan."""
service = ScanService(test_db)
service = ScanService(db)
result = service.get_scan(999)
assert result is None
def test_get_scan_found(self, test_db, sample_config_file):
def test_get_scan_found(self, db, sample_db_config):
"""Test getting an existing scan."""
service = ScanService(test_db)
service = ScanService(db)
# Create a scan
scan_id = service.trigger_scan(sample_config_file)
scan_id = service.trigger_scan(config_id=sample_db_config.id)
# Retrieve it
result = service.get_scan(scan_id)
@@ -90,9 +83,9 @@ class TestScanServiceGet:
class TestScanServiceList:
"""Tests for listing scans."""
def test_list_scans_empty(self, test_db):
def test_list_scans_empty(self, db):
"""Test listing scans when database is empty."""
service = ScanService(test_db)
service = ScanService(db)
result = service.list_scans(page=1, per_page=20)
@@ -100,13 +93,13 @@ class TestScanServiceList:
assert len(result.items) == 0
assert result.pages == 0
def test_list_scans_with_data(self, test_db, sample_config_file):
def test_list_scans_with_data(self, db, sample_db_config):
"""Test listing scans with multiple scans."""
service = ScanService(test_db)
service = ScanService(db)
# Create 3 scans
for i in range(3):
service.trigger_scan(sample_config_file, triggered_by='api')
service.trigger_scan(config_id=sample_db_config.id, triggered_by='api')
# List all scans
result = service.list_scans(page=1, per_page=20)
@@ -115,13 +108,13 @@ class TestScanServiceList:
assert len(result.items) == 3
assert result.pages == 1
def test_list_scans_pagination(self, test_db, sample_config_file):
def test_list_scans_pagination(self, db, sample_db_config):
"""Test pagination."""
service = ScanService(test_db)
service = ScanService(db)
# Create 5 scans
for i in range(5):
service.trigger_scan(sample_config_file)
service.trigger_scan(config_id=sample_db_config.id)
# Get page 1 (2 items per page)
result = service.list_scans(page=1, per_page=2)
@@ -141,18 +134,18 @@ class TestScanServiceList:
assert len(result.items) == 1
assert result.has_next is False
def test_list_scans_filter_by_status(self, test_db, sample_config_file):
def test_list_scans_filter_by_status(self, db, sample_db_config):
"""Test filtering scans by status."""
service = ScanService(test_db)
service = ScanService(db)
# Create scans with different statuses
scan_id_1 = service.trigger_scan(sample_config_file)
scan_id_2 = service.trigger_scan(sample_config_file)
scan_id_1 = service.trigger_scan(config_id=sample_db_config.id)
scan_id_2 = service.trigger_scan(config_id=sample_db_config.id)
# Mark one as completed
scan = test_db.query(Scan).filter(Scan.id == scan_id_1).first()
scan = db.query(Scan).filter(Scan.id == scan_id_1).first()
scan.status = 'completed'
test_db.commit()
db.commit()
# Filter by running
result = service.list_scans(status_filter='running')
@@ -162,9 +155,9 @@ class TestScanServiceList:
result = service.list_scans(status_filter='completed')
assert result.total == 1
def test_list_scans_invalid_status_filter(self, test_db):
def test_list_scans_invalid_status_filter(self, db):
"""Test filtering with invalid status."""
service = ScanService(test_db)
service = ScanService(db)
with pytest.raises(ValueError, match="Invalid status"):
service.list_scans(status_filter='invalid_status')
@@ -173,46 +166,46 @@ class TestScanServiceList:
class TestScanServiceDelete:
"""Tests for deleting scans."""
def test_delete_scan_not_found(self, test_db):
def test_delete_scan_not_found(self, db):
"""Test deleting a nonexistent scan."""
service = ScanService(test_db)
service = ScanService(db)
with pytest.raises(ValueError, match="not found"):
service.delete_scan(999)
def test_delete_scan_success(self, test_db, sample_config_file):
def test_delete_scan_success(self, db, sample_db_config):
"""Test successful scan deletion."""
service = ScanService(test_db)
service = ScanService(db)
# Create a scan
scan_id = service.trigger_scan(sample_config_file)
scan_id = service.trigger_scan(config_id=sample_db_config.id)
# Verify it exists
assert test_db.query(Scan).filter(Scan.id == scan_id).first() is not None
assert db.query(Scan).filter(Scan.id == scan_id).first() is not None
# Delete it
result = service.delete_scan(scan_id)
assert result is True
# Verify it's gone
assert test_db.query(Scan).filter(Scan.id == scan_id).first() is None
assert db.query(Scan).filter(Scan.id == scan_id).first() is None
class TestScanServiceStatus:
"""Tests for scan status retrieval."""
def test_get_scan_status_not_found(self, test_db):
def test_get_scan_status_not_found(self, db):
"""Test getting status of nonexistent scan."""
service = ScanService(test_db)
service = ScanService(db)
result = service.get_scan_status(999)
assert result is None
def test_get_scan_status_running(self, test_db, sample_config_file):
def test_get_scan_status_running(self, db, sample_db_config):
"""Test getting status of running scan."""
service = ScanService(test_db)
service = ScanService(db)
scan_id = service.trigger_scan(sample_config_file)
scan_id = service.trigger_scan(config_id=sample_db_config.id)
status = service.get_scan_status(scan_id)
assert status is not None
@@ -221,16 +214,16 @@ class TestScanServiceStatus:
assert status['progress'] == 'In progress'
assert status['title'] == 'Test Scan'
def test_get_scan_status_completed(self, test_db, sample_config_file):
def test_get_scan_status_completed(self, db, sample_db_config):
"""Test getting status of completed scan."""
service = ScanService(test_db)
service = ScanService(db)
# Create and mark as completed
scan_id = service.trigger_scan(sample_config_file)
scan = test_db.query(Scan).filter(Scan.id == scan_id).first()
scan_id = service.trigger_scan(config_id=sample_db_config.id)
scan = db.query(Scan).filter(Scan.id == scan_id).first()
scan.status = 'completed'
scan.duration = 125.5
test_db.commit()
db.commit()
status = service.get_scan_status(scan_id)
@@ -242,35 +235,35 @@ class TestScanServiceStatus:
class TestScanServiceDatabaseMapping:
"""Tests for mapping scan reports to database models."""
def test_save_scan_to_db(self, test_db, sample_config_file, sample_scan_report):
def test_save_scan_to_db(self, db, sample_db_config, sample_scan_report):
"""Test saving a complete scan report to database."""
service = ScanService(test_db)
service = ScanService(db)
# Create a scan
scan_id = service.trigger_scan(sample_config_file)
scan_id = service.trigger_scan(config_id=sample_db_config.id)
# Save report to database
service._save_scan_to_db(sample_scan_report, scan_id, status='completed')
# Verify scan updated
scan = test_db.query(Scan).filter(Scan.id == scan_id).first()
scan = db.query(Scan).filter(Scan.id == scan_id).first()
assert scan.status == 'completed'
assert scan.duration == 125.5
# Verify sites created
sites = test_db.query(ScanSite).filter(ScanSite.scan_id == scan_id).all()
sites = db.query(ScanSite).filter(ScanSite.scan_id == scan_id).all()
assert len(sites) == 1
assert sites[0].site_name == 'Test Site'
# Verify IPs created
ips = test_db.query(ScanIP).filter(ScanIP.scan_id == scan_id).all()
ips = db.query(ScanIP).filter(ScanIP.scan_id == scan_id).all()
assert len(ips) == 1
assert ips[0].ip_address == '192.168.1.10'
assert ips[0].ping_expected is True
assert ips[0].ping_actual is True
# Verify ports created (TCP: 22, 80, 443, 8080 | UDP: 53)
ports = test_db.query(ScanPort).filter(ScanPort.scan_id == scan_id).all()
ports = db.query(ScanPort).filter(ScanPort.scan_id == scan_id).all()
assert len(ports) == 5 # 4 TCP + 1 UDP
# Verify TCP ports
@@ -285,7 +278,7 @@ class TestScanServiceDatabaseMapping:
assert udp_ports[0].port == 53
# Verify services created
services = test_db.query(ScanServiceModel).filter(
services = db.query(ScanServiceModel).filter(
ScanServiceModel.scan_id == scan_id
).all()
assert len(services) == 4 # SSH, HTTP (80), HTTPS, HTTP (8080)
@@ -300,15 +293,15 @@ class TestScanServiceDatabaseMapping:
assert https_service.http_protocol == 'https'
assert https_service.screenshot_path == 'screenshots/192_168_1_10_443.png'
def test_map_port_expected_vs_actual(self, test_db, sample_config_file, sample_scan_report):
def test_map_port_expected_vs_actual(self, db, sample_db_config, sample_scan_report):
"""Test that expected vs actual ports are correctly flagged."""
service = ScanService(test_db)
service = ScanService(db)
scan_id = service.trigger_scan(sample_config_file)
scan_id = service.trigger_scan(config_id=sample_db_config.id)
service._save_scan_to_db(sample_scan_report, scan_id)
# Check TCP ports
tcp_ports = test_db.query(ScanPort).filter(
tcp_ports = db.query(ScanPort).filter(
ScanPort.scan_id == scan_id,
ScanPort.protocol == 'tcp'
).all()
@@ -322,15 +315,15 @@ class TestScanServiceDatabaseMapping:
# Port 8080 was not expected
assert port.expected is False, f"Port {port.port} should not be expected"
def test_map_certificate_and_tls(self, test_db, sample_config_file, sample_scan_report):
def test_map_certificate_and_tls(self, db, sample_db_config, sample_scan_report):
"""Test that certificate and TLS data are correctly mapped."""
service = ScanService(test_db)
service = ScanService(db)
scan_id = service.trigger_scan(sample_config_file)
scan_id = service.trigger_scan(config_id=sample_db_config.id)
service._save_scan_to_db(sample_scan_report, scan_id)
# Find HTTPS service
https_service = test_db.query(ScanServiceModel).filter(
https_service = db.query(ScanServiceModel).filter(
ScanServiceModel.scan_id == scan_id,
ScanServiceModel.service_name == 'https'
).first()
@@ -363,11 +356,11 @@ class TestScanServiceDatabaseMapping:
assert tls_13 is not None
assert tls_13.supported is True
def test_get_scan_with_full_details(self, test_db, sample_config_file, sample_scan_report):
def test_get_scan_with_full_details(self, db, sample_db_config, sample_scan_report):
"""Test retrieving scan with all nested relationships."""
service = ScanService(test_db)
service = ScanService(db)
scan_id = service.trigger_scan(sample_config_file)
scan_id = service.trigger_scan(config_id=sample_db_config.id)
service._save_scan_to_db(sample_scan_report, scan_id)
# Get full scan details