302 lines
11 KiB
Python
302 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Standalone test script for DEER Power Outages API.
|
|
|
|
This script tests the DEER API endpoints without any Home Assistant dependencies.
|
|
Usage: python test_api.py [REGION] [CITY] [STREET]
|
|
Example: python test_api.py TN Cluj-Napoca "Strada Avram Iancu"
|
|
"""
|
|
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
import sys
|
|
import re
|
|
from typing import Optional
|
|
|
|
|
|
API_BASE_URL = "https://outages.distributie-energie.ro/api"
|
|
API_INCIDENTS = "/incidents/{region}/0"
|
|
API_SCHEDULED_15 = "/scheduled/{region}/0/15"
|
|
API_SCHEDULED_TODAY = "/scheduled/{region}/0/azi"
|
|
|
|
|
|
def strip_html_tags(text: str) -> str:
|
|
"""Remove HTML tags from text."""
|
|
if not text:
|
|
return text
|
|
return re.sub(r'<[^>]+>', ' ', text).strip()
|
|
|
|
|
|
def matches_location_planned(item: dict, city: str, street: Optional[str]) -> bool:
|
|
"""Check if planned outage matches configured location."""
|
|
fields_to_check = [
|
|
item.get("ansambluFunctional", ""),
|
|
item.get("adreseLucrari", ""),
|
|
]
|
|
|
|
# Combine all fields to search
|
|
combined = " ".join(str(f) for f in fields_to_check if f).lower()
|
|
|
|
# Check city (required)
|
|
if city.lower() not in combined:
|
|
return False
|
|
|
|
# Check street (optional)
|
|
if street and street.strip():
|
|
if street.lower() not in combined:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def matches_location_unplanned(item: dict, city: str, street: Optional[str]) -> bool:
|
|
"""Check if unplanned outage matches configured location."""
|
|
adresa = item.get("adresa", "").lower()
|
|
|
|
# Check city (required)
|
|
if city.lower() not in adresa:
|
|
return False
|
|
|
|
# Check street (optional)
|
|
if street and street.strip():
|
|
if street.lower() not in adresa:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
async def test_incidents(session: aiohttp.ClientSession, region: str, city: Optional[str] = None, street: Optional[str] = None):
|
|
"""Test unplanned incidents endpoint."""
|
|
url = API_BASE_URL + API_INCIDENTS.format(region=region)
|
|
|
|
print(f"\n{'='*80}")
|
|
print(f"Testing Incidents (Unplanned) - Region: {region}")
|
|
print(f"URL: {url}")
|
|
print(f"{'='*80}")
|
|
|
|
try:
|
|
async with asyncio.timeout(30):
|
|
async with session.get(url) as response:
|
|
print(f"Status: {response.status}")
|
|
response.raise_for_status()
|
|
data = await response.json()
|
|
|
|
print(f"Total incidents: {len(data)}")
|
|
|
|
if city:
|
|
filtered = [item for item in data if matches_location_unplanned(item, city, street)]
|
|
print(f"Filtered incidents for {city}" + (f", {street}" if street else "") + f": {len(filtered)}")
|
|
data = filtered
|
|
|
|
if data:
|
|
print(f"\nFirst {min(3, len(data))} incidents:")
|
|
for i, item in enumerate(data[:3], 1):
|
|
print(f"\n--- Incident {i} ---")
|
|
print(f"ID: {item.get('id')}")
|
|
print(f"County: {item.get('judet')}")
|
|
print(f"Address: {strip_html_tags(item.get('adresa', ''))}")
|
|
print(f"Start: {item.get('dataStart')}")
|
|
print(f"End: {item.get('dataStop')}")
|
|
else:
|
|
print("No incidents found" + (f" for {city}" if city else ""))
|
|
|
|
except asyncio.TimeoutError:
|
|
print("ERROR: Timeout connecting to API")
|
|
except aiohttp.ClientError as err:
|
|
print(f"ERROR: HTTP error - {err}")
|
|
except json.JSONDecodeError:
|
|
print("ERROR: Invalid JSON response")
|
|
except Exception as err:
|
|
print(f"ERROR: Unexpected error - {err}")
|
|
|
|
|
|
async def test_scheduled_15(session: aiohttp.ClientSession, region: str, city: Optional[str] = None, street: Optional[str] = None):
|
|
"""Test scheduled outages (15 days) endpoint."""
|
|
url = API_BASE_URL + API_SCHEDULED_15.format(region=region)
|
|
|
|
print(f"\n{'='*80}")
|
|
print(f"Testing Scheduled Outages (15 days) - Region: {region}")
|
|
print(f"URL: {url}")
|
|
print(f"{'='*80}")
|
|
|
|
try:
|
|
async with asyncio.timeout(30):
|
|
async with session.get(url) as response:
|
|
print(f"Status: {response.status}")
|
|
response.raise_for_status()
|
|
data = await response.json()
|
|
|
|
print(f"Total scheduled outages: {len(data)}")
|
|
|
|
if city:
|
|
filtered = [item for item in data if matches_location_planned(item, city, street)]
|
|
print(f"Filtered outages for {city}" + (f", {street}" if street else "") + f": {len(filtered)}")
|
|
data = filtered
|
|
|
|
if data:
|
|
print(f"\nFirst {min(3, len(data))} scheduled outages:")
|
|
for i, item in enumerate(data[:3], 1):
|
|
print(f"\n--- Scheduled Outage {i} ---")
|
|
print(f"ID: {item.get('id')}")
|
|
print(f"Branch: {item.get('sucursala')}")
|
|
print(f"Location: {item.get('ansambluFunctional')}")
|
|
print(f"Work Address: {item.get('adreseLucrari')}")
|
|
print(f"Start: {item.get('dataProgramareStart')}")
|
|
print(f"End: {item.get('dataProgramareStop')}")
|
|
print(f"Duration: {item.get('durataProgramare')}")
|
|
else:
|
|
print("No scheduled outages found" + (f" for {city}" if city else ""))
|
|
|
|
except asyncio.TimeoutError:
|
|
print("ERROR: Timeout connecting to API")
|
|
except aiohttp.ClientError as err:
|
|
print(f"ERROR: HTTP error - {err}")
|
|
except json.JSONDecodeError:
|
|
print("ERROR: Invalid JSON response")
|
|
except Exception as err:
|
|
print(f"ERROR: Unexpected error - {err}")
|
|
|
|
|
|
async def test_scheduled_today(session: aiohttp.ClientSession, region: str, city: Optional[str] = None, street: Optional[str] = None):
|
|
"""Test scheduled outages (today) endpoint."""
|
|
url = API_BASE_URL + API_SCHEDULED_TODAY.format(region=region)
|
|
|
|
print(f"\n{'='*80}")
|
|
print(f"Testing Scheduled Outages (Today) - Region: {region}")
|
|
print(f"URL: {url}")
|
|
print(f"{'='*80}")
|
|
|
|
try:
|
|
async with asyncio.timeout(30):
|
|
async with session.get(url) as response:
|
|
print(f"Status: {response.status}")
|
|
response.raise_for_status()
|
|
data = await response.json()
|
|
|
|
print(f"Total today's outages: {len(data)}")
|
|
|
|
if city:
|
|
filtered = [item for item in data if matches_location_planned(item, city, street)]
|
|
print(f"Filtered outages for {city}" + (f", {street}" if street else "") + f": {len(filtered)}")
|
|
data = filtered
|
|
|
|
if data:
|
|
print(f"\nAll today's scheduled outages:")
|
|
for i, item in enumerate(data, 1):
|
|
print(f"\n--- Today's Outage {i} ---")
|
|
print(f"ID: {item.get('id')}")
|
|
print(f"Branch: {item.get('sucursala')}")
|
|
print(f"Location: {item.get('ansambluFunctional')}")
|
|
print(f"Work Address: {item.get('adreseLucrari')}")
|
|
print(f"Start: {item.get('dataProgramareStart')}")
|
|
print(f"End: {item.get('dataProgramareStop')}")
|
|
print(f"Duration: {item.get('durataProgramare')}")
|
|
else:
|
|
print("No outages scheduled for today" + (f" for {city}" if city else ""))
|
|
|
|
except asyncio.TimeoutError:
|
|
print("ERROR: Timeout connecting to API")
|
|
except aiohttp.ClientError as err:
|
|
print(f"ERROR: HTTP error - {err}")
|
|
except json.JSONDecodeError:
|
|
print("ERROR: Invalid JSON response")
|
|
except Exception as err:
|
|
print(f"ERROR: Unexpected error - {err}")
|
|
|
|
|
|
async def main(region: str = "TN", city: Optional[str] = None, street: Optional[str] = None):
|
|
"""Main test function."""
|
|
print(f"\n{'#'*80}")
|
|
print(f"# DEER Power Outages API Test")
|
|
print(f"# Region: {region}")
|
|
if city:
|
|
print(f"# City: {city}")
|
|
if street:
|
|
print(f"# Street: {street}")
|
|
print(f"{'#'*80}")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
# Test all three endpoints
|
|
await test_incidents(session, region, city, street)
|
|
await test_scheduled_15(session, region, city, street)
|
|
await test_scheduled_today(session, region, city, street)
|
|
|
|
print(f"\n{'#'*80}")
|
|
print("# Test completed")
|
|
print(f"{'#'*80}\n")
|
|
|
|
|
|
def test_filtering_logic():
|
|
"""Test location filtering logic with sample data."""
|
|
print(f"\n{'='*80}")
|
|
print("Testing Filtering Logic")
|
|
print(f"{'='*80}\n")
|
|
|
|
# Sample planned outage data
|
|
planned_sample = {
|
|
"id": "82540",
|
|
"ansambluFunctional": "CALINESTI, Strada PRINCIPALA",
|
|
"adreseLucrari": "Zona Centrala",
|
|
"dataProgramareStart": "14/01/2026 08:00",
|
|
"dataProgramareStop": "14/01/2026 15:00",
|
|
"durataProgramare": "7h : 0'"
|
|
}
|
|
|
|
# Sample unplanned incident data
|
|
unplanned_sample = {
|
|
"id": "69706",
|
|
"adresa": "Loc: SURA MARE, Strada: VETERANILOR<br />Loc: SURA MARE",
|
|
"dataStart": "10/01/2026 01:00",
|
|
"dataStop": ""
|
|
}
|
|
|
|
# Test cases
|
|
test_cases = [
|
|
("Planned - Match city only", planned_sample, "CALINESTI", None, matches_location_planned, True),
|
|
("Planned - Match city (case insensitive)", planned_sample, "calinesti", None, matches_location_planned, True),
|
|
("Planned - Match city and street", planned_sample, "CALINESTI", "PRINCIPALA", matches_location_planned, True),
|
|
("Planned - City match, wrong street", planned_sample, "CALINESTI", "SECUNDARA", matches_location_planned, False),
|
|
("Planned - Wrong city", planned_sample, "CLUJ", None, matches_location_planned, False),
|
|
("Unplanned - Match city only", unplanned_sample, "SURA MARE", None, matches_location_unplanned, True),
|
|
("Unplanned - Match city (case insensitive)", unplanned_sample, "sura mare", None, matches_location_unplanned, True),
|
|
("Unplanned - Match city and street", unplanned_sample, "SURA MARE", "VETERANILOR", matches_location_unplanned, True),
|
|
("Unplanned - City match, wrong street", unplanned_sample, "SURA MARE", "CENTRALA", matches_location_unplanned, False),
|
|
("Unplanned - Wrong city", unplanned_sample, "CLUJ", None, matches_location_unplanned, False),
|
|
]
|
|
|
|
passed = 0
|
|
failed = 0
|
|
|
|
for test_name, data, city, street, func, expected in test_cases:
|
|
result = func(data, city, street)
|
|
status = "✓ PASS" if result == expected else "✗ FAIL"
|
|
if result == expected:
|
|
passed += 1
|
|
else:
|
|
failed += 1
|
|
print(f"{status}: {test_name}")
|
|
if result != expected:
|
|
print(f" Expected: {expected}, Got: {result}")
|
|
|
|
print(f"\n{'='*80}")
|
|
print(f"Results: {passed} passed, {failed} failed")
|
|
print(f"{'='*80}\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Parse command line arguments
|
|
region = sys.argv[1] if len(sys.argv) > 1 else "TN"
|
|
city = sys.argv[2] if len(sys.argv) > 2 else None
|
|
street = sys.argv[3] if len(sys.argv) > 3 else None
|
|
|
|
# Validate region
|
|
if region not in ["TN", "TS", "MN"]:
|
|
print(f"ERROR: Invalid region '{region}'. Must be one of: TN, TS, MN")
|
|
sys.exit(1)
|
|
|
|
# Run filtering logic tests
|
|
test_filtering_logic()
|
|
|
|
# Run API tests
|
|
asyncio.run(main(region, city, street))
|