209 lines
8.2 KiB
Python
209 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test Collaboration Matching Integration
|
|
========================================
|
|
|
|
This script tests the collaboration matching functionality by:
|
|
1. Creating two IT audits with matching criteria (Proxmox PBS + open_to_backup_replication)
|
|
2. Running the collaboration matching algorithm
|
|
3. Verifying matches appear in the database
|
|
|
|
Usage:
|
|
# From the worktree directory with DEV database running:
|
|
# Set DATABASE_URL environment variable with proper credentials, then run:
|
|
DATABASE_URL=postgresql://nordabiz_app:YOUR_PASSWORD@localhost:5433/nordabiz \
|
|
python3 scripts/test_collaboration_matching.py
|
|
|
|
Requirements:
|
|
- PostgreSQL DEV database running on localhost:5433
|
|
- IT audit tables created (run migration first)
|
|
|
|
Author: Norda Biznes Development Team
|
|
Created: 2026-01-09
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
# Set database URL for DEV environment
|
|
# WARNING: The fallback DATABASE_URL uses a placeholder password.
|
|
# Production credentials MUST be set via the DATABASE_URL environment variable.
|
|
# NEVER commit real credentials to version control (CWE-798).
|
|
if 'DATABASE_URL' not in os.environ:
|
|
os.environ['DATABASE_URL'] = 'postgresql://nordabiz_app:CHANGE_ME@localhost:5433/nordabiz'
|
|
|
|
# Import after setting DATABASE_URL
|
|
from database import SessionLocal, Company, ITAudit, ITCollaborationMatch
|
|
from it_audit_service import ITAuditService
|
|
|
|
|
|
def get_test_companies(db, limit=2):
|
|
"""Get two test companies from the database"""
|
|
companies = db.query(Company).filter(
|
|
Company.status == 'active'
|
|
).limit(limit).all()
|
|
return companies
|
|
|
|
|
|
def create_test_audit(db, company_id: int, audit_data: dict) -> ITAudit:
|
|
"""Create a test IT audit for a company"""
|
|
service = ITAuditService(db)
|
|
return service.save_audit(company_id, audit_data)
|
|
|
|
|
|
def run_matching_test():
|
|
"""Run the collaboration matching test"""
|
|
print("=" * 60)
|
|
print("IT Audit Collaboration Matching Test")
|
|
print("=" * 60)
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
# Step 1: Get two test companies
|
|
print("\n[1] Getting test companies...")
|
|
companies = get_test_companies(db, limit=2)
|
|
|
|
if len(companies) < 2:
|
|
print("ERROR: Need at least 2 active companies in database")
|
|
return False
|
|
|
|
company_a, company_b = companies[0], companies[1]
|
|
print(f" Company A: {company_a.name} (ID: {company_a.id})")
|
|
print(f" Company B: {company_b.name} (ID: {company_b.id})")
|
|
|
|
# Step 2: Check for existing audits and delete them for clean test
|
|
print("\n[2] Cleaning up existing test audits...")
|
|
db.query(ITAudit).filter(
|
|
ITAudit.company_id.in_([company_a.id, company_b.id])
|
|
).delete(synchronize_session='fetch')
|
|
|
|
db.query(ITCollaborationMatch).filter(
|
|
(ITCollaborationMatch.company_a_id.in_([company_a.id, company_b.id])) |
|
|
(ITCollaborationMatch.company_b_id.in_([company_a.id, company_b.id]))
|
|
).delete(synchronize_session='fetch')
|
|
|
|
db.commit()
|
|
print(" Existing audits and matches cleaned up.")
|
|
|
|
# Step 3: Create audit for Company A with PBS + open_to_backup_replication
|
|
print("\n[3] Creating IT audit for Company A with Proxmox PBS...")
|
|
audit_a_data = {
|
|
'has_proxmox_pbs': True,
|
|
'open_to_backup_replication': True,
|
|
'backup_solution': 'Proxmox Backup Server',
|
|
'backup_frequency': 'daily',
|
|
'has_edr': True,
|
|
'has_mfa': True,
|
|
'has_azure_ad': True,
|
|
'has_m365': True,
|
|
'open_to_shared_licensing': True,
|
|
'open_to_teams_federation': True,
|
|
'employee_count': '11-50',
|
|
'monitoring_solution': 'Zabbix',
|
|
'open_to_shared_monitoring': True,
|
|
}
|
|
audit_a = create_test_audit(db, company_a.id, audit_a_data)
|
|
print(f" Audit A created: ID={audit_a.id}, PBS={audit_a.has_proxmox_pbs}, "
|
|
f"Open to backup replication={audit_a.open_to_backup_replication}")
|
|
print(f" Scores: overall={audit_a.overall_score}, security={audit_a.security_score}, "
|
|
f"collaboration={audit_a.collaboration_score}")
|
|
|
|
# Step 4: Create audit for Company B with PBS + open_to_backup_replication
|
|
print("\n[4] Creating IT audit for Company B with Proxmox PBS...")
|
|
audit_b_data = {
|
|
'has_proxmox_pbs': True,
|
|
'open_to_backup_replication': True,
|
|
'backup_solution': 'Proxmox Backup Server',
|
|
'backup_frequency': 'daily',
|
|
'has_edr': True,
|
|
'has_mfa': True,
|
|
'has_azure_ad': True,
|
|
'has_m365': True,
|
|
'open_to_shared_licensing': True,
|
|
'open_to_teams_federation': True,
|
|
'employee_count': '51-100',
|
|
'monitoring_solution': 'Zabbix',
|
|
'open_to_shared_monitoring': True,
|
|
}
|
|
audit_b = create_test_audit(db, company_b.id, audit_b_data)
|
|
print(f" Audit B created: ID={audit_b.id}, PBS={audit_b.has_proxmox_pbs}, "
|
|
f"Open to backup replication={audit_b.open_to_backup_replication}")
|
|
print(f" Scores: overall={audit_b.overall_score}, security={audit_b.security_score}, "
|
|
f"collaboration={audit_b.collaboration_score}")
|
|
|
|
# Step 5: Run collaboration matching for Company A
|
|
print("\n[5] Running collaboration matching for Company A...")
|
|
service = ITAuditService(db)
|
|
matches = service.find_collaboration_matches(company_a.id)
|
|
print(f" Found {len(matches)} potential matches")
|
|
|
|
# Step 6: Save matches to database
|
|
print("\n[6] Saving matches to database...")
|
|
for match in matches:
|
|
saved_match = service.save_collaboration_match(match)
|
|
print(f" Saved: {match.company_a_name} <-> {match.company_b_name}")
|
|
print(f" Type: {match.match_type}, Score: {match.match_score}")
|
|
print(f" Reason: {match.match_reason}")
|
|
|
|
# Step 7: Verify matches in database
|
|
print("\n[7] Verifying matches in database...")
|
|
db_matches = db.query(ITCollaborationMatch).filter(
|
|
(ITCollaborationMatch.company_a_id == company_a.id) |
|
|
(ITCollaborationMatch.company_b_id == company_a.id)
|
|
).all()
|
|
|
|
print(f" Total matches in DB: {len(db_matches)}")
|
|
|
|
# Check for backup_replication match
|
|
backup_matches = [m for m in db_matches if m.match_type == 'backup_replication']
|
|
has_backup_match = len(backup_matches) > 0
|
|
|
|
if has_backup_match:
|
|
print("\n✅ SUCCESS: Backup replication match found!")
|
|
bm = backup_matches[0]
|
|
print(f" Company A ID: {bm.company_a_id}")
|
|
print(f" Company B ID: {bm.company_b_id}")
|
|
print(f" Match Score: {bm.match_score}")
|
|
print(f" Status: {bm.status}")
|
|
else:
|
|
print("\n❌ FAILED: No backup replication match found!")
|
|
|
|
# Print all match types found
|
|
print("\n[8] All match types found:")
|
|
for match in db_matches:
|
|
print(f" - {match.match_type}: {match.company_a_id} <-> {match.company_b_id} (score: {match.match_score})")
|
|
|
|
# Summary
|
|
print("\n" + "=" * 60)
|
|
print("TEST SUMMARY")
|
|
print("=" * 60)
|
|
print(f"Companies tested: {company_a.name}, {company_b.name}")
|
|
print(f"Audits created: 2")
|
|
print(f"Matches found: {len(db_matches)}")
|
|
print(f"Backup replication match: {'YES ✅' if has_backup_match else 'NO ❌'}")
|
|
|
|
# Expected matches based on test data
|
|
expected_match_types = ['backup_replication', 'shared_licensing', 'teams_federation', 'shared_monitoring']
|
|
found_types = {m.match_type for m in db_matches}
|
|
missing_types = set(expected_match_types) - found_types
|
|
|
|
if missing_types:
|
|
print(f"\nMissing expected match types: {missing_types}")
|
|
|
|
return has_backup_match
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ ERROR: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
success = run_matching_test()
|
|
sys.exit(0 if success else 1)
|