0655b4c4f8
- Added customer_id and contact_id filters to the list_opportunities endpoint for improved querying. - Implemented a redirect for opportunity detail pages to a new format. - Refactored SubscriptionMatrixService to load invoices from a local snapshot instead of an external service, improving performance and reliability. - Updated settings to include 'pipeline' as a case type and added a new section for managing ignored product texts in the invoice error finder. - Introduced a new user_sag_create_preferences table to store per-user default case types for new cases. - Enhanced frontend settings page with invoice error finder configuration options and improved handling of ignored product texts. - Added migrations to support new features, including resolved status for invoice error finder issues and user-specific case type preferences.
708 lines
25 KiB
Python
708 lines
25 KiB
Python
import sys
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from main import app
|
|
|
|
|
|
def test_internet_connections_module_routes_are_available():
|
|
client = TestClient(app)
|
|
|
|
health_response = client.get('/api/v1/internet-connections/health')
|
|
assert health_response.status_code == 200
|
|
assert health_response.json()['service'] == 'internet-connections-module'
|
|
|
|
page_response = client.get('/economy/internet-connections')
|
|
assert page_response.status_code == 200
|
|
assert (
|
|
'Internetforbindelser' in page_response.text
|
|
or "window.location.href = '/login'" in page_response.text
|
|
)
|
|
|
|
detail_response = client.get('/economy/internet-connections/1')
|
|
assert detail_response.status_code == 200
|
|
assert (
|
|
'IP-ranges' in detail_response.text
|
|
or "window.location.href = '/login'" in detail_response.text
|
|
)
|
|
|
|
|
|
def test_create_ip_range_rejects_invalid_cidr(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.post('/api/v1/internet-connections/2/ip-ranges', json={
|
|
'name': 'LAN',
|
|
'cidr': 'not-a-cidr',
|
|
'description': 'Test range',
|
|
})
|
|
|
|
assert response.status_code == 400
|
|
assert 'CIDR' in response.json()['detail']
|
|
|
|
|
|
def test_create_connection_requires_address(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', lambda query, params=None: [])
|
|
|
|
client = TestClient(app)
|
|
response = client.post('/api/v1/internet-connections', json={
|
|
'name': 'Uden adresse',
|
|
'provider': 'GlobalConnect A/S',
|
|
'status': 'active',
|
|
'allocation_model': 'shared',
|
|
'value_type': 'bmc_networks',
|
|
})
|
|
|
|
assert response.status_code == 400
|
|
assert 'address is required' in response.json()['detail']
|
|
|
|
|
|
def test_list_ip_ranges_includes_ipam_details(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'FROM internet_connections_ip_ranges' in query:
|
|
return [{
|
|
'id': 9,
|
|
'connection_id': 2,
|
|
'name': 'LAN',
|
|
'cidr': '10.0.0.0/24',
|
|
'description': 'Test range',
|
|
}]
|
|
if 'FROM internet_connections_ip_addresses' in query:
|
|
return [{
|
|
'id': 1,
|
|
'range_id': 9,
|
|
'ip_address': '10.0.0.1',
|
|
'status': 'in_use',
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections/2/ip-ranges')
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()[0]
|
|
assert payload['network_address'] == '10.0.0.0'
|
|
assert payload['usable_hosts'] == 254
|
|
assert payload['used_addresses'] == 1
|
|
|
|
|
|
def test_migration_wizard_v2_query_requires_real_segment_match(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if "FROM internet_connections_customer_documents" in query:
|
|
return [{
|
|
"id": 101,
|
|
"customer_id": 77,
|
|
"connection_id": None,
|
|
"original_filename": "management_net.txt",
|
|
"filename": "management_net.txt",
|
|
"file_size": 1024,
|
|
"mime_type": "text/plain",
|
|
"extracted_text": "1 UNTAGGED Native Management\nVlan 50 Not in Use\nIP Informationer",
|
|
"notes": None,
|
|
"created_at": None,
|
|
}]
|
|
if "FROM internet_connections_customer_document_segments" in query:
|
|
return [{
|
|
"id": 201,
|
|
"document_id": 101,
|
|
"block_index": 0,
|
|
"block_title": "IP Informationer",
|
|
"content": "1 UNTAGGED Native Management\nVlan 50 Not in Use\nIP Informationer",
|
|
"ip_addresses": [],
|
|
"cidr_blocks": [],
|
|
"references_json": [],
|
|
"socket_numbers": [],
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, "execute_query", fake_execute_query)
|
|
monkeypatch.setattr(internet_router, "_ensure_document_segments", lambda document_id, extracted_text: 1)
|
|
|
|
payload = asyncio.run(internet_router._build_customer_document_hits(77, "Karise", "stageone"))
|
|
|
|
assert payload["segments"] == []
|
|
assert payload["documents"] == []
|
|
|
|
|
|
def test_migration_wizard_v2_query_keeps_precise_segment_hits(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if "FROM internet_connections_customer_documents" in query:
|
|
return [{
|
|
"id": 102,
|
|
"customer_id": 77,
|
|
"connection_id": None,
|
|
"original_filename": "karise.txt",
|
|
"filename": "karise.txt",
|
|
"file_size": 2048,
|
|
"mime_type": "text/plain",
|
|
"extracted_text": "StageOne uplink til Karise\nPort 1 StageOne WAN U20",
|
|
"notes": None,
|
|
"created_at": None,
|
|
}]
|
|
if "FROM internet_connections_customer_document_segments" in query:
|
|
return [{
|
|
"id": 202,
|
|
"document_id": 102,
|
|
"block_index": 0,
|
|
"block_title": "StageOne uplink til Karise",
|
|
"content": "StageOne uplink til Karise\nPort 1 StageOne WAN U20",
|
|
"ip_addresses": [],
|
|
"cidr_blocks": [],
|
|
"references_json": [],
|
|
"socket_numbers": ["U20"],
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, "execute_query", fake_execute_query)
|
|
monkeypatch.setattr(internet_router, "_ensure_document_segments", lambda document_id, extracted_text: 1)
|
|
|
|
payload = asyncio.run(internet_router._build_customer_document_hits(77, "Karise", "stageone"))
|
|
|
|
assert len(payload["segments"]) == 1
|
|
assert payload["segments"][0]["title"] == "StageOne uplink til Karise"
|
|
assert payload["documents"][0]["snippet_count"] == 1
|
|
|
|
|
|
def test_migration_wizard_v2_returns_full_text_for_selected_segment(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
monkeypatch.setattr(internet_router, "execute_query_single", lambda query, params: {
|
|
"segment_id": 202,
|
|
"document_id": 102,
|
|
"block_index": 3,
|
|
"title": "StageOne uplink",
|
|
"original_filename": "karise.txt",
|
|
"content": "Hele den valgte tekstblok\nmed alle linjer.",
|
|
})
|
|
|
|
payload = asyncio.run(internet_router.get_customer_document_segment(202))
|
|
|
|
assert payload["title"] == "StageOne uplink"
|
|
assert payload["content"] == "Hele den valgte tekstblok\nmed alle linjer."
|
|
|
|
|
|
def test_migration_wizard_block_search_requires_all_words(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if "FROM internet_connections_customer_documents" in query:
|
|
return [{
|
|
"id": 103, "customer_id": 77, "connection_id": None,
|
|
"original_filename": "sales.txt", "filename": "sales.txt",
|
|
"file_size": 1, "mime_type": "text/plain", "notes": None,
|
|
"created_at": None, "extracted_text": "Management network notes",
|
|
}]
|
|
if "FROM internet_connections_customer_document_segments" in query:
|
|
return [{
|
|
"id": 203, "document_id": 103, "block_index": 0,
|
|
"block_title": "Management", "content": "Management network notes",
|
|
"ip_addresses": [], "cidr_blocks": [], "references_json": [], "socket_numbers": [],
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, "execute_query", fake_execute_query)
|
|
monkeypatch.setattr(internet_router, "_ensure_document_segments", lambda document_id, extracted_text: 1)
|
|
|
|
payload = asyncio.run(internet_router._build_customer_document_hits(77, "Karise", "sales management"))
|
|
|
|
assert payload["segments"] == []
|
|
assert payload["documents"] == []
|
|
|
|
|
|
def test_create_ip_range_auto_generates_addresses_from_cidr(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
created_addresses = []
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'INSERT INTO internet_connections_ip_ranges' in query:
|
|
return [{
|
|
'id': 15,
|
|
'connection_id': 2,
|
|
'name': 'LAN',
|
|
'cidr': '192.168.1.0/30',
|
|
'description': 'Auto generated',
|
|
}]
|
|
if 'INSERT INTO internet_connections_ip_addresses' in query:
|
|
created_addresses.append(params[1])
|
|
return [{
|
|
'id': len(created_addresses),
|
|
'range_id': 15,
|
|
'ip_address': params[1],
|
|
'status': 'available',
|
|
}]
|
|
if 'INSERT INTO internet_connections_history' in query:
|
|
return [{
|
|
'id': 20,
|
|
'connection_id': 2,
|
|
'event_type': 'ip_range_created',
|
|
'summary': 'Created IP range',
|
|
'details': {},
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.post('/api/v1/internet-connections/2/ip-ranges', json={
|
|
'name': 'LAN',
|
|
'cidr': '192.168.1.0/30',
|
|
'description': 'Auto generated',
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
assert len(created_addresses) == 2
|
|
assert created_addresses == ['192.168.1.1', '192.168.1.2']
|
|
|
|
|
|
def test_create_ip_range_skips_duplicate_existing_ip_addresses(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
created_addresses = []
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'INSERT INTO internet_connections_ip_ranges' in query:
|
|
return [{
|
|
'id': 15,
|
|
'connection_id': 2,
|
|
'name': 'LAN',
|
|
'cidr': '192.168.1.0/30',
|
|
'description': 'Auto generated',
|
|
}]
|
|
if 'FROM internet_connections_ip_addresses' in query and 'WHERE ip_address = %s' in query:
|
|
if params[0] == '192.168.1.1':
|
|
return [{'id': 99, 'ip_address': '192.168.1.1'}]
|
|
return []
|
|
if 'INSERT INTO internet_connections_ip_addresses' in query:
|
|
created_addresses.append(params[1])
|
|
return [{
|
|
'id': len(created_addresses),
|
|
'range_id': 15,
|
|
'ip_address': params[1],
|
|
'status': 'available',
|
|
}]
|
|
if 'INSERT INTO internet_connections_history' in query:
|
|
return [{'id': 20}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.post('/api/v1/internet-connections/2/ip-ranges', json={
|
|
'name': 'LAN',
|
|
'cidr': '192.168.1.0/30',
|
|
'description': 'Auto generated',
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
assert created_addresses == ['192.168.1.2']
|
|
|
|
|
|
def test_update_ip_address_status_changes_record(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
updated = []
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'UPDATE internet_connections_ip_addresses' in query:
|
|
updated.append(params)
|
|
return [{'id': 4, 'status': 'reserved'}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.put('/api/v1/internet-connections/2/ip-addresses/4', json={
|
|
'status': 'reserved',
|
|
'assigned_to': 'Test device',
|
|
'comment': 'Reserved for switch',
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
assert updated[0][0] == 'reserved'
|
|
assert updated[0][1] == 'Test device'
|
|
|
|
|
|
def test_create_ip_address_rejects_duplicate_ip(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'FROM internet_connections_ip_addresses' in query and 'WHERE ip_address = %s' in query:
|
|
return [{'id': 4, 'range_id': 2}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.post('/api/v1/internet-connections/2/ip-addresses', json={
|
|
'range_id': 7,
|
|
'ip_address': '10.0.0.10',
|
|
'status': 'available',
|
|
})
|
|
|
|
assert response.status_code == 409
|
|
assert 'findes allerede' in response.json()['detail']
|
|
|
|
|
|
def test_list_ip_addresses_returns_structured_payload(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'FROM internet_connections_ip_addresses' in query and 'JOIN internet_connections_ip_ranges' in query:
|
|
return [{
|
|
'id': 7,
|
|
'range_id': 3,
|
|
'ip_address': '10.0.0.10',
|
|
'status': 'in_use',
|
|
'assigned_to': 'Router',
|
|
'assigned_type': 'device',
|
|
'comment': 'Main gateway',
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections/2/ip-addresses')
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()[0]
|
|
assert payload['status_label'] == 'I brug'
|
|
assert payload['badge_class'] == 'bg-primary'
|
|
|
|
|
|
def test_create_ip_range_writes_history(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
calls = []
|
|
|
|
def fake_execute_query(query, params=None):
|
|
calls.append((query, params))
|
|
if 'INSERT INTO internet_connections_ip_ranges' in query:
|
|
return [{
|
|
'id': 1,
|
|
'connection_id': 2,
|
|
'name': 'LAN',
|
|
'cidr': '10.0.0.0/24',
|
|
'description': 'Test range',
|
|
}]
|
|
if 'INSERT INTO internet_connections_history' in query:
|
|
return [{
|
|
'id': 11,
|
|
'connection_id': 2,
|
|
'event_type': 'ip_range_created',
|
|
'summary': 'Created IP range',
|
|
'details': {},
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.post('/api/v1/internet-connections/2/ip-ranges', json={
|
|
'name': 'LAN',
|
|
'cidr': '10.0.0.0/24',
|
|
'description': 'Test range',
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
assert any('INSERT INTO internet_connections_history' in query for query, _ in calls)
|
|
|
|
|
|
def test_ip_address_status_summary_endpoint(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'COUNT(*)' in query and 'internet_connections_ip_addresses' in query:
|
|
return [{
|
|
'available': 2,
|
|
'in_use': 1,
|
|
'reserved': 1,
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections/2/ip-addresses/summary')
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()['available'] == 2
|
|
assert response.json()['in_use'] == 1
|
|
|
|
|
|
def test_pricing_history_can_be_created(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'INSERT INTO internet_connections_pricing' in query:
|
|
return [{
|
|
'id': 7,
|
|
'connection_id': 2,
|
|
'effective_from': '2026-07-01',
|
|
'purchase_price': 1500,
|
|
'sales_price': 1800,
|
|
'notes': 'Ny aftale',
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.post('/api/v1/internet-connections/2/pricing', json={
|
|
'effective_from': '2026-07-01',
|
|
'purchase_price': 1500,
|
|
'sales_price': 1800,
|
|
'notes': 'Ny aftale',
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()['sales_price'] == 1800
|
|
|
|
|
|
def test_contract_overview_endpoint_reports_status(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'FROM internet_connections_connections' in query and 'contract_end' in query:
|
|
return [{
|
|
'id': 3,
|
|
'name': 'Test connection',
|
|
'provider': 'BMC',
|
|
'contract_start': '2025-01-01',
|
|
'contract_end': '2025-12-31',
|
|
'status': 'active',
|
|
'sales_price': 900,
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections/contracts')
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()[0]['contract_status'] == 'expired'
|
|
|
|
|
|
def test_contract_overview_endpoint_supports_status_filter(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'status = %s' in query:
|
|
return [{
|
|
'id': 4,
|
|
'name': 'Filtered contract',
|
|
'provider': 'Nordic',
|
|
'contract_start': '2026-01-01',
|
|
'contract_end': '2026-12-31',
|
|
'status': 'active',
|
|
'sales_price': 1200,
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections/contracts', params={'status': 'active'})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()[0]['status'] == 'active'
|
|
|
|
|
|
def test_contract_overview_page_contains_table_controls():
|
|
client = TestClient(app)
|
|
response = client.get('/economy/internet-connections')
|
|
|
|
assert response.status_code == 200
|
|
assert (
|
|
'connectionsTableBody' in response.text
|
|
or "window.location.href = '/login'" in response.text
|
|
)
|
|
assert (
|
|
'pageSummaryText' in response.text
|
|
or "window.location.href = '/login'" in response.text
|
|
)
|
|
|
|
|
|
def test_contract_overview_returns_empty_list_when_query_fails(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fail_execute_query(query, params=None):
|
|
raise RuntimeError('database unavailable')
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fail_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections/contracts')
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
|
|
def test_list_connections_returns_empty_list_when_query_fails(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fail_execute_query(query, params=None):
|
|
raise RuntimeError('database unavailable')
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fail_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections')
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
|
|
def test_pricing_summary_returns_zeroes_when_query_fails(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fail_execute_query(query, params=None):
|
|
raise RuntimeError('database unavailable')
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fail_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections/pricing/summary')
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
'total_connections': 0,
|
|
'active_connections': 0,
|
|
'shared_head_connections': 0,
|
|
'total_purchase_cost': 0,
|
|
'total_sales_price': 0,
|
|
'total_margin': 0,
|
|
}
|
|
|
|
|
|
def test_create_connection_requires_subscription_id_for_subscription_value(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', lambda query, params=None: [])
|
|
|
|
client = TestClient(app)
|
|
response = client.post('/api/v1/internet-connections', json={
|
|
'name': 'Shared transit',
|
|
'address': 'Testvej 1, 8000 Aarhus C',
|
|
'allocation_model': 'shared',
|
|
'value_type': 'subscription',
|
|
})
|
|
|
|
assert response.status_code == 400
|
|
assert 'subscription_id' in response.json()['detail']
|
|
|
|
|
|
def test_create_connection_requires_value_label_for_other(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', lambda query, params=None: [])
|
|
|
|
client = TestClient(app)
|
|
response = client.post('/api/v1/internet-connections', json={
|
|
'name': 'Carrier edge',
|
|
'address': 'Testvej 1, 8000 Aarhus C',
|
|
'allocation_model': 'dedicated',
|
|
'value_type': 'other',
|
|
})
|
|
|
|
assert response.status_code == 400
|
|
assert 'value_label' in response.json()['detail']
|
|
|
|
|
|
def test_list_connections_supports_shared_only_filter(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
assert "ic.allocation_model = 'shared' AND ic.parent_id IS NULL" in query
|
|
return [{
|
|
'id': 10,
|
|
'parent_id': None,
|
|
'name': 'BMC Networks · NKA008225',
|
|
'provider': 'GlobalConnect',
|
|
'customer_id': 1662,
|
|
'customer_name': 'BMC Networks',
|
|
'parent_name': None,
|
|
'address': None,
|
|
'status': 'active',
|
|
'monthly_cost': 256,
|
|
'sales_price': 0,
|
|
'margin_amount': -256,
|
|
'technology': 'Fiber',
|
|
'connection_type': 'fiber',
|
|
'circuit_number': 'NKA008225',
|
|
'speed_mbps': 100,
|
|
'upload_mbps': 100,
|
|
'download_mbps': 100,
|
|
'monitoring_url': None,
|
|
'contract_start': None,
|
|
'contract_end': None,
|
|
'allocation_model': 'shared',
|
|
'value_type': 'bmc_networks',
|
|
'value_label': None,
|
|
'subscription_id': None,
|
|
'subscription_number': None,
|
|
'subscription_product_name': None,
|
|
'subscription_customer_name': None,
|
|
'ip_range_count': 3,
|
|
'total_ip_addresses': 62,
|
|
'in_use_ip_addresses': 0,
|
|
'reserved_ip_addresses': 0,
|
|
'available_ip_addresses': 62,
|
|
}]
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections', params={'shared_only': 'true'})
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()[0]
|
|
assert payload['allocation_model_label'] == 'Delt'
|
|
assert payload['value_type_label'] == 'BMC Networks'
|
|
assert payload['is_shared_head'] is True
|
|
|
|
|
|
def test_subscription_options_endpoint_returns_lookup_rows(monkeypatch):
|
|
from app.modules.internet_connections.backend import router as internet_router
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if 'FROM sag_subscriptions s' in query:
|
|
return [{
|
|
'id': 9,
|
|
'subscription_number': 'SUB-1001',
|
|
'product_name': 'Internet 1G',
|
|
'customer_id': 1662,
|
|
'customer_name': 'BMC Networks',
|
|
'status': 'active',
|
|
}]
|
|
return []
|
|
|
|
monkeypatch.setattr(internet_router, 'execute_query', fake_execute_query)
|
|
|
|
client = TestClient(app)
|
|
response = client.get('/api/v1/internet-connections/subscription-options', params={'q': '1G'})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()[0]['subscription_number'] == 'SUB-1001'
|