113 lines
3.9 KiB
Python
113 lines
3.9 KiB
Python
"""Unit tests for data collectors"""
|
|
|
|
import pytest
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from app.services.collectors.top500 import TOP500Collector
|
|
from app.services.collectors.base import BaseCollector, HTTPCollector
|
|
|
|
|
|
class TestBaseCollector:
|
|
"""Tests for BaseCollector"""
|
|
|
|
def test_base_collector_attributes(self):
|
|
"""Test base collector has correct default attributes via concrete class"""
|
|
collector = TOP500Collector()
|
|
assert collector.name == "top500"
|
|
assert collector.priority == "P0"
|
|
assert collector.module == "L1"
|
|
assert collector.frequency_hours == 4
|
|
|
|
|
|
class TestTOP500Collector:
|
|
"""Tests for TOP500Collector"""
|
|
|
|
def test_parse_coordinate_valid_float(self):
|
|
"""Test parsing valid float coordinate"""
|
|
collector = TOP500Collector()
|
|
assert collector._parse_coordinate(45.5) == 45.5
|
|
|
|
def test_parse_coordinate_valid_string(self):
|
|
"""Test parsing valid string coordinate"""
|
|
collector = TOP500Collector()
|
|
assert collector._parse_coordinate("45.5") == 45.5
|
|
|
|
def test_parse_coordinate_invalid_string(self):
|
|
"""Test parsing invalid string coordinate"""
|
|
collector = TOP500Collector()
|
|
assert collector._parse_coordinate("invalid") == 0.0
|
|
|
|
def test_parse_coordinate_none(self):
|
|
"""Test parsing None coordinate"""
|
|
collector = TOP500Collector()
|
|
assert collector._parse_coordinate(None) == 0.0
|
|
|
|
def test_parse_response_empty(self):
|
|
"""Test parsing empty response"""
|
|
collector = TOP500Collector()
|
|
result = collector.parse_response({"items": []})
|
|
assert result == []
|
|
|
|
def test_parse_response_single_item(self):
|
|
"""Test parsing single item response"""
|
|
collector = TOP500Collector()
|
|
response = {
|
|
"items": [
|
|
{
|
|
"rank": 1,
|
|
"system_name": "Test Supercomputer",
|
|
"country": "USA",
|
|
"city": "San Francisco",
|
|
"latitude": 37.7749,
|
|
"longitude": -122.4194,
|
|
"manufacturer": "Test Corp",
|
|
"r_max": 100000.0,
|
|
"r_peak": 150000.0,
|
|
"power": 5000.0,
|
|
"cores": 100000,
|
|
"interconnect": "InfiniBand",
|
|
"os": "Linux",
|
|
}
|
|
]
|
|
}
|
|
result = collector.parse_response(response)
|
|
assert len(result) == 1
|
|
assert result[0]["cluster_id"] == "top500_1"
|
|
assert result[0]["name"] == "Test Supercomputer"
|
|
assert result[0]["country"] == "USA"
|
|
assert result[0]["rank"] == 1
|
|
assert result[0]["source"] == "TOP500"
|
|
|
|
def test_parse_response_skips_invalid_item(self):
|
|
"""Test parsing skips items with missing data"""
|
|
collector = TOP500Collector()
|
|
response = {
|
|
"items": [
|
|
{"rank": 1, "system_name": "Valid"},
|
|
{"rank": None, "system_name": "Invalid"},
|
|
]
|
|
}
|
|
result = collector.parse_response(response)
|
|
assert len(result) == 1
|
|
assert result[0]["name"] == "Valid"
|
|
|
|
|
|
class TestHTTPCollector:
|
|
"""Tests for HTTPCollector"""
|
|
|
|
def test_http_collector_attributes(self):
|
|
"""Test HTTP collector has correct default attributes via concrete class"""
|
|
collector = TOP500Collector()
|
|
assert collector.base_url == "https://top500.org/api/v1.0/lists/"
|
|
assert collector.name == "top500"
|
|
assert collector.priority == "P0"
|
|
|
|
def test_collector_has_required_methods(self):
|
|
"""Test HTTP collector has required methods"""
|
|
collector = TOP500Collector()
|
|
assert hasattr(collector, "fetch")
|
|
assert hasattr(collector, "parse_response")
|
|
assert callable(collector.fetch)
|
|
assert callable(collector.parse_response)
|