590 lines
20 KiB
Python
590 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
End-to-End Tests für EspoCRM Custom Entities
|
|
Automatisierte Tests für CRUD-Operationen und Relationships
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
from typing import Dict, List, Optional, Set
|
|
from datetime import datetime, date
|
|
from dataclasses import dataclass, field
|
|
|
|
from espocrm_api_client import EspoCRMAPIClient
|
|
|
|
|
|
# ============================================================================
|
|
# Configuration
|
|
# ============================================================================
|
|
|
|
CONFIG = {
|
|
'base_url': 'https://crm.bitbylaw.com',
|
|
'api_key': '2b0747ca34d15032aa233ae043cc61bc',
|
|
'username': 'dev-test'
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# Test Results Tracking
|
|
# ============================================================================
|
|
|
|
@dataclass
|
|
class TestResult:
|
|
"""Track test execution results"""
|
|
entity_type: str
|
|
test_name: str
|
|
success: bool
|
|
duration: float
|
|
error: Optional[str] = None
|
|
details: Dict = field(default_factory=dict)
|
|
|
|
|
|
class TestTracker:
|
|
"""Track and report test results"""
|
|
|
|
def __init__(self):
|
|
self.results: List[TestResult] = []
|
|
self.created_records: Dict[str, List[str]] = {} # entity_type -> [ids]
|
|
|
|
def add_result(self, result: TestResult):
|
|
"""Add test result"""
|
|
self.results.append(result)
|
|
|
|
def track_created(self, entity_type: str, record_id: str):
|
|
"""Track created record for cleanup"""
|
|
if entity_type not in self.created_records:
|
|
self.created_records[entity_type] = []
|
|
self.created_records[entity_type].append(record_id)
|
|
|
|
def print_summary(self):
|
|
"""Print test summary"""
|
|
total = len(self.results)
|
|
passed = sum(1 for r in self.results if r.success)
|
|
failed = total - passed
|
|
total_time = sum(r.duration for r in self.results)
|
|
|
|
print("\n" + "=" * 80)
|
|
print("TEST SUMMARY".center(80))
|
|
print("=" * 80)
|
|
|
|
# Group by entity
|
|
by_entity = {}
|
|
for result in self.results:
|
|
if result.entity_type not in by_entity:
|
|
by_entity[result.entity_type] = []
|
|
by_entity[result.entity_type].append(result)
|
|
|
|
for entity_type, results in sorted(by_entity.items()):
|
|
entity_passed = sum(1 for r in results if r.success)
|
|
entity_total = len(results)
|
|
permission_errors = sum(1 for r in results if not r.success and '[PERMISSION]' in (r.error or ''))
|
|
|
|
if entity_passed == entity_total:
|
|
status = "✅"
|
|
elif permission_errors > 0:
|
|
status = "🔒" # Locked due to permissions
|
|
else:
|
|
status = "⚠️"
|
|
|
|
print(f"\n{status} {entity_type}: {entity_passed}/{entity_total} tests passed")
|
|
if permission_errors > 0:
|
|
print(f" (⚠️ {permission_errors} test(s) skipped due to missing permissions)")
|
|
|
|
for result in results:
|
|
icon = "✓" if result.success else "✗"
|
|
time_str = f"{result.duration:.3f}s"
|
|
print(f" {icon} {result.test_name:<40} {time_str:>8}")
|
|
|
|
if not result.success and result.error:
|
|
# Don't print full error for permission issues
|
|
if '[PERMISSION]' not in result.error:
|
|
print(f" Error: {result.error}")
|
|
|
|
print("\n" + "=" * 80)
|
|
print(f"Total: {passed}/{total} tests passed ({failed} failed)")
|
|
print(f"Time: {total_time:.2f}s")
|
|
print("=" * 80 + "\n")
|
|
|
|
return failed == 0
|
|
|
|
|
|
# ============================================================================
|
|
# Base Test Class
|
|
# ============================================================================
|
|
|
|
class EntityTestBase:
|
|
"""Base class for entity tests"""
|
|
|
|
def __init__(self, client: EspoCRMAPIClient, tracker: TestTracker):
|
|
self.client = client
|
|
self.tracker = tracker
|
|
self.entity_type = None # To be set by subclass
|
|
|
|
def run_test(self, test_name: str, test_func):
|
|
"""Run a single test with timing and error handling"""
|
|
start_time = time.time()
|
|
|
|
try:
|
|
result = test_func()
|
|
duration = time.time() - start_time
|
|
|
|
self.tracker.add_result(TestResult(
|
|
entity_type=self.entity_type,
|
|
test_name=test_name,
|
|
success=True,
|
|
duration=duration,
|
|
details=result or {}
|
|
))
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
duration = time.time() - start_time
|
|
error_msg = str(e)
|
|
|
|
# Check if it's a permission error (403)
|
|
is_permission_error = '403' in error_msg or 'Forbidden' in error_msg
|
|
|
|
self.tracker.add_result(TestResult(
|
|
entity_type=self.entity_type,
|
|
test_name=test_name,
|
|
success=False,
|
|
duration=duration,
|
|
error=f"{'[PERMISSION] ' if is_permission_error else ''}{error_msg}"
|
|
))
|
|
|
|
if is_permission_error:
|
|
print(f"⚠️ {self.entity_type}.{test_name} skipped: No permission")
|
|
else:
|
|
print(f"❌ {self.entity_type}.{test_name} failed: {error_msg}")
|
|
return None
|
|
|
|
def test_create(self, data: Dict) -> Optional[str]:
|
|
"""Test create operation"""
|
|
def _test():
|
|
record = self.client.create(self.entity_type, data)
|
|
record_id = record.get('id')
|
|
assert record_id, "No ID returned"
|
|
|
|
self.tracker.track_created(self.entity_type, record_id)
|
|
print(f"✓ Created {self.entity_type}: {record_id}")
|
|
|
|
return record_id
|
|
|
|
return self.run_test('create', _test)
|
|
|
|
def test_read(self, record_id: str) -> Optional[Dict]:
|
|
"""Test read operation"""
|
|
def _test():
|
|
record = self.client.read(self.entity_type, record_id)
|
|
assert record.get('id') == record_id, "ID mismatch"
|
|
|
|
print(f"✓ Read {self.entity_type}: {record_id}")
|
|
return record
|
|
|
|
return self.run_test('read', _test)
|
|
|
|
def test_update(self, record_id: str, data: Dict) -> Optional[Dict]:
|
|
"""Test update operation"""
|
|
def _test():
|
|
record = self.client.update(self.entity_type, record_id, data)
|
|
assert record.get('id') == record_id, "ID mismatch"
|
|
|
|
print(f"✓ Updated {self.entity_type}: {record_id}")
|
|
return record
|
|
|
|
return self.run_test('update', _test)
|
|
|
|
def test_delete(self, record_id: str) -> bool:
|
|
"""Test delete operation"""
|
|
def _test():
|
|
self.client.delete(self.entity_type, record_id)
|
|
print(f"✓ Deleted {self.entity_type}: {record_id}")
|
|
return True
|
|
|
|
return self.run_test('delete', _test)
|
|
|
|
def test_link(self, record_id: str, link_name: str, foreign_id: str) -> bool:
|
|
"""Test relationship link"""
|
|
def _test():
|
|
self.client.link(self.entity_type, record_id, link_name, foreign_id)
|
|
|
|
# Verify link
|
|
linked = self.client.get_linked(self.entity_type, record_id, link_name)
|
|
linked_ids = [r['id'] for r in linked]
|
|
assert foreign_id in linked_ids, f"Link not found: {foreign_id}"
|
|
|
|
print(f"✓ Linked {self.entity_type}:{record_id} -> {link_name}:{foreign_id}")
|
|
return True
|
|
|
|
return self.run_test(f'link_{link_name}', _test)
|
|
|
|
def test_unlink(self, record_id: str, link_name: str, foreign_id: str) -> bool:
|
|
"""Test relationship unlink"""
|
|
def _test():
|
|
self.client.unlink(self.entity_type, record_id, link_name, foreign_id)
|
|
|
|
# Verify unlink
|
|
linked = self.client.get_linked(self.entity_type, record_id, link_name)
|
|
linked_ids = [r['id'] for r in linked]
|
|
assert foreign_id not in linked_ids, f"Link still exists: {foreign_id}"
|
|
|
|
print(f"✓ Unlinked {self.entity_type}:{record_id} -x- {link_name}:{foreign_id}")
|
|
return True
|
|
|
|
return self.run_test(f'unlink_{link_name}', _test)
|
|
|
|
|
|
# ============================================================================
|
|
# Specific Entity Tests
|
|
# ============================================================================
|
|
|
|
class CMietobjektTest(EntityTestBase):
|
|
"""Tests for CMietobjekt entity"""
|
|
|
|
def __init__(self, client: EspoCRMAPIClient, tracker: TestTracker):
|
|
super().__init__(client, tracker)
|
|
self.entity_type = 'CMietobjekt'
|
|
|
|
def run_full_test(self) -> Optional[str]:
|
|
"""Run complete CRUD test"""
|
|
# Create
|
|
data = {
|
|
'name': f'Test Mietobjekt {datetime.now().strftime("%Y%m%d_%H%M%S")}',
|
|
'objekttyp': 'Wohnung',
|
|
'lage': 'Teststraße 123, 12345 Teststadt',
|
|
'anschriftStreet': 'Teststraße 123',
|
|
'anschriftCity': 'Teststadt',
|
|
'anschriftPostalCode': '12345',
|
|
'anschriftCountry': 'Deutschland'
|
|
}
|
|
|
|
record_id = self.test_create(data)
|
|
if not record_id:
|
|
return None
|
|
|
|
# Read
|
|
record = self.test_read(record_id)
|
|
if not record:
|
|
return None
|
|
|
|
# Update
|
|
update_data = {'lage': 'Neue Teststraße 456, 54321 Neustadt'}
|
|
self.test_update(record_id, update_data)
|
|
|
|
return record_id
|
|
|
|
|
|
class CVmhMietverhaeltnisTest(EntityTestBase):
|
|
"""Tests for CVmhMietverhältnis entity"""
|
|
|
|
def __init__(self, client: EspoCRMAPIClient, tracker: TestTracker):
|
|
super().__init__(client, tracker)
|
|
self.entity_type = 'CVmhMietverhltnis'
|
|
|
|
def run_full_test(self, mietobjekt_id: Optional[str] = None) -> Optional[str]:
|
|
"""Run complete CRUD test"""
|
|
# Create
|
|
data = {
|
|
'name': f'Test Mietverhältnis {datetime.now().strftime("%Y%m%d_%H%M%S")}',
|
|
'nutzungsart': 'Wohnraum',
|
|
'beendigungsTatbestand': 'Kündigung Vermieter',
|
|
'status': 'Bestehend',
|
|
'kaltmiete': 850.00,
|
|
'warmmiete': 1050.00,
|
|
'bKPauschale': 150.00,
|
|
'bKVorauszahlung': 50.00,
|
|
'vertragsdatum': date.today().isoformat(),
|
|
'auszugsfrist': '2026-06-30'
|
|
}
|
|
|
|
if mietobjekt_id:
|
|
data['vmhMietobjektId'] = mietobjekt_id
|
|
|
|
record_id = self.test_create(data)
|
|
if not record_id:
|
|
return None
|
|
|
|
# Read
|
|
record = self.test_read(record_id)
|
|
if not record:
|
|
return None
|
|
|
|
# Update
|
|
update_data = {'kaltmiete': 900.00, 'warmmiete': 1100.00}
|
|
self.test_update(record_id, update_data)
|
|
|
|
return record_id
|
|
|
|
|
|
class CKuendigungTest(EntityTestBase):
|
|
"""Tests for CKündigung entity"""
|
|
|
|
def __init__(self, client: EspoCRMAPIClient, tracker: TestTracker):
|
|
super().__init__(client, tracker)
|
|
self.entity_type = 'CKuendigung'
|
|
|
|
def run_full_test(self) -> Optional[str]:
|
|
"""Run complete CRUD test"""
|
|
# Create
|
|
data = {
|
|
'name': f'Test Kündigung {datetime.now().strftime("%Y%m%d_%H%M%S")}',
|
|
'beendigungsTatbestand': 'Kündigung Vermieter'
|
|
}
|
|
|
|
record_id = self.test_create(data)
|
|
if not record_id:
|
|
return None
|
|
|
|
# Read
|
|
record = self.test_read(record_id)
|
|
if not record:
|
|
return None
|
|
|
|
# Update
|
|
update_data = {'status': 'Versendet', 'gegenstandswert': 5000.00}
|
|
self.test_update(record_id, update_data)
|
|
|
|
return record_id
|
|
|
|
|
|
class CBeteiligteTest(EntityTestBase):
|
|
"""Tests for CBeteiligte entity"""
|
|
|
|
def __init__(self, client: EspoCRMAPIClient, tracker: TestTracker):
|
|
super().__init__(client, tracker)
|
|
self.entity_type = 'CBeteiligte'
|
|
|
|
def run_full_test(self) -> Optional[str]:
|
|
"""Run complete CRUD test"""
|
|
# Create
|
|
data = {
|
|
'lastName': f'Testperson_{datetime.now().strftime("%H%M%S")}',
|
|
'firstName': 'Max',
|
|
'salutationName': 'Mr.',
|
|
'addressStreet': 'Musterstraße 1',
|
|
'addressCity': 'Musterstadt',
|
|
'addressPostalCode': '12345',
|
|
'addressCountry': 'Deutschland',
|
|
'emailAddress': f'test_{datetime.now().strftime("%Y%m%d%H%M%S")}@example.com',
|
|
'phoneNumber': '+49 123 456789'
|
|
}
|
|
|
|
record_id = self.test_create(data)
|
|
if not record_id:
|
|
return None
|
|
|
|
# Read
|
|
record = self.test_read(record_id)
|
|
if not record:
|
|
return None
|
|
|
|
# Update
|
|
update_data = {'phoneNumber': '+49 987 654321'}
|
|
self.test_update(record_id, update_data)
|
|
|
|
return record_id
|
|
|
|
|
|
class CMietinkassoTest(EntityTestBase):
|
|
"""Tests for CMietinkasso entity"""
|
|
|
|
def __init__(self, client: EspoCRMAPIClient, tracker: TestTracker):
|
|
super().__init__(client, tracker)
|
|
self.entity_type = 'CMietinkasso'
|
|
|
|
def run_full_test(self) -> Optional[str]:
|
|
"""Run complete CRUD test"""
|
|
# Create
|
|
data = {
|
|
'name': f'Test Mietinkasso {datetime.now().strftime("%Y%m%d_%H%M%S")}',
|
|
'gegenstandswert': 7500.00,
|
|
'syncStatus': 'clean'
|
|
}
|
|
|
|
record_id = self.test_create(data)
|
|
if not record_id:
|
|
return None
|
|
|
|
# Read
|
|
self.test_read(record_id)
|
|
|
|
# Update
|
|
update_data = {'gegenstandswert': 8000.00}
|
|
self.test_update(record_id, update_data)
|
|
|
|
return record_id
|
|
|
|
|
|
class CVmhRaeumungsklageTest(EntityTestBase):
|
|
"""Tests for CVmhRäumungsklage entity"""
|
|
|
|
def __init__(self, client: EspoCRMAPIClient, tracker: TestTracker):
|
|
super().__init__(client, tracker)
|
|
self.entity_type = 'CVmhRumungsklage'
|
|
|
|
def run_full_test(self) -> Optional[str]:
|
|
"""Run complete CRUD test"""
|
|
# Create
|
|
data = {
|
|
'name': f'Test Räumungsklage {datetime.now().strftime("%Y%m%d_%H%M%S")}',
|
|
'gegenstandswert': 12000.00,
|
|
'syncStatus': 'clean'
|
|
}
|
|
|
|
record_id = self.test_create(data)
|
|
if not record_id:
|
|
return None
|
|
|
|
# Read
|
|
self.test_read(record_id)
|
|
|
|
# Update
|
|
update_data = {'gegenstandswert': 13000.00}
|
|
self.test_update(record_id, update_data)
|
|
|
|
return record_id
|
|
|
|
|
|
# ============================================================================
|
|
# Main Test Runner
|
|
# ============================================================================
|
|
|
|
def run_all_tests():
|
|
"""Run all E2E tests"""
|
|
print("=" * 80)
|
|
print("ESPOCRM E2E TESTS".center(80))
|
|
print("=" * 80)
|
|
print(f"\nTarget: {CONFIG['base_url']}")
|
|
print(f"User: {CONFIG['username']}\n")
|
|
|
|
# Initialize
|
|
client = EspoCRMAPIClient(
|
|
base_url=CONFIG['base_url'],
|
|
api_key=CONFIG['api_key'],
|
|
username=CONFIG['username']
|
|
)
|
|
|
|
tracker = TestTracker()
|
|
|
|
# Test connection
|
|
print("Testing API connection...")
|
|
if not client.check_connection():
|
|
print("❌ Connection failed. Aborting tests.")
|
|
return False
|
|
print("✓ Connection successful\n")
|
|
|
|
print("=" * 80)
|
|
print("RUNNING TESTS".center(80))
|
|
print("=" * 80 + "\n")
|
|
|
|
try:
|
|
# ========================================================================
|
|
# Basic Entity CRUD Tests
|
|
# ========================================================================
|
|
|
|
print("🔷 Testing CMietobjekt...")
|
|
mietobjekt_test = CMietobjektTest(client, tracker)
|
|
mietobjekt_id = mietobjekt_test.run_full_test()
|
|
|
|
print("\n🔷 Testing CVmhMietverhältnis...")
|
|
mietverhaeltnis_test = CVmhMietverhaeltnisTest(client, tracker)
|
|
mietverhaeltnis_id = mietverhaeltnis_test.run_full_test(mietobjekt_id)
|
|
|
|
print("\n🔷 Testing CKündigung...")
|
|
kuendigung_test = CKuendigungTest(client, tracker)
|
|
kuendigung_id = kuendigung_test.run_full_test()
|
|
|
|
print("\n🔷 Testing CBeteiligte...")
|
|
beteiligte_test = CBeteiligteTest(client, tracker)
|
|
beteiligte_id = beteiligte_test.run_full_test()
|
|
|
|
print("\n🔷 Testing CMietinkasso...")
|
|
mietinkasso_test = CMietinkassoTest(client, tracker)
|
|
mietinkasso_id = mietinkasso_test.run_full_test()
|
|
|
|
print("\n🔷 Testing CVmhRäumungsklage...")
|
|
raeumungsklage_test = CVmhRaeumungsklageTest(client, tracker)
|
|
raeumungsklage_id = raeumungsklage_test.run_full_test()
|
|
|
|
# ========================================================================
|
|
# Relationship Tests
|
|
# ========================================================================
|
|
|
|
if mietverhaeltnis_id and mietobjekt_id:
|
|
print("\n🔗 Testing Relationships...")
|
|
|
|
# Test Mietverhältnis -> Mietobjekt (already linked via vmhMietobjektId)
|
|
# We can verify it was linked correctly
|
|
linked_mietobjekt = client.read('CVmhMietverhltnis', mietverhaeltnis_id)
|
|
if linked_mietobjekt.get('vmhMietobjektId') == mietobjekt_id:
|
|
print(f"✓ CVmhMietverhältnis linked to CMietobjekt via vmhMietobjektId")
|
|
|
|
# Note: Some hasMany relationships may use different API patterns
|
|
# and are not tested here to avoid false negatives
|
|
|
|
# if mietverhaeltnis_id and beteiligte_id:
|
|
# # Link Beteiligte as Mieter
|
|
# mietverhaeltnis_test.test_link(
|
|
# mietverhaeltnis_id,
|
|
# 'vmhbeteiligtemieter',
|
|
# beteiligte_id
|
|
# )
|
|
#
|
|
# # Unlink
|
|
# mietverhaeltnis_test.test_unlink(
|
|
# mietverhaeltnis_id,
|
|
# 'vmhbeteiligtemieter',
|
|
# beteiligte_id
|
|
# )
|
|
|
|
# ========================================================================
|
|
# Cleanup
|
|
# ========================================================================
|
|
|
|
print("\n🧹 Cleaning up test data...")
|
|
|
|
# Delete in reverse order (respect dependencies)
|
|
deletion_order = [
|
|
'CKuendigung',
|
|
'CMietinkasso',
|
|
'CVmhRumungsklage',
|
|
'CVmhMietverhltnis',
|
|
'CBeteiligte',
|
|
'CMietobjekt'
|
|
]
|
|
|
|
for entity_type in deletion_order:
|
|
if entity_type in tracker.created_records:
|
|
for record_id in tracker.created_records[entity_type]:
|
|
try:
|
|
client.delete(entity_type, record_id)
|
|
print(f"✓ Deleted {entity_type}: {record_id}")
|
|
except Exception as e:
|
|
print(f"⚠️ Could not delete {entity_type}:{record_id}: {e}")
|
|
|
|
print("\n✓ Cleanup complete")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n⚠️ Tests interrupted by user")
|
|
return False
|
|
except Exception as e:
|
|
print(f"\n\n❌ Unexpected error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
# Print summary
|
|
success = tracker.print_summary()
|
|
|
|
return success
|
|
|
|
|
|
# ============================================================================
|
|
# Entry Point
|
|
# ============================================================================
|
|
|
|
if __name__ == '__main__':
|
|
success = run_all_tests()
|
|
sys.exit(0 if success else 1)
|