mirror of
https://framagit.org/framasoft/framaspace/argos.git
synced 2025-04-28 18:02:41 +02:00
Alexis Métaireau
43f8aabb2c
- Restructured server module to separate the application creation and configuration. - Moved code dealing with SQLAlchemy database setup and teardown to the main application file. - Moved functions related to configuration file loading to `argos.server.settings`. - Fixed SQLAchemy expressions in `argos.server.queries`. - Implemented a more granular system of setting checks' schedule on the server. - Introduced frequency scheduling on per-website basis in the YAML config. - Introduced Pytest fixtures for handling test database and authorized HTTP client in `tests/conftest.py`. - Included a first test for the api - Implemented changes to models to accommodate changes to task scheduling. - Fixed errors concerning database concurrency arising from changes to the application setup.
107 lines
2.5 KiB
Python
107 lines
2.5 KiB
Python
from typing import Dict, List, Literal, Optional, Tuple
|
|
|
|
from pydantic import BaseModel, HttpUrl, field_validator
|
|
from pydantic.functional_validators import BeforeValidator
|
|
from typing_extensions import Annotated
|
|
|
|
from argos.schemas.utils import string_to_duration
|
|
|
|
# This file contains the pydantic schemas.
|
|
# For the database models, check in argos.server.models.
|
|
|
|
|
|
Severity = Literal["warning", "error", "critical"]
|
|
|
|
|
|
def parse_threshold(value):
|
|
for duration_str, severity in value.items():
|
|
days = string_to_duration(duration_str, "days")
|
|
# Return here because it's one-item dicts.
|
|
return (days, severity)
|
|
|
|
|
|
class SSL(BaseModel):
|
|
thresholds: List[Annotated[Tuple[int, Severity], BeforeValidator(parse_threshold)]]
|
|
|
|
|
|
class WebsiteCheck(BaseModel):
|
|
key: str
|
|
value: str | List[str] | Dict[str, str]
|
|
|
|
class Config:
|
|
arbitrary_types_allowed = True
|
|
|
|
@classmethod
|
|
def __get_validators__(cls):
|
|
yield cls.validate
|
|
|
|
@classmethod
|
|
def validate(cls, value):
|
|
if isinstance(value, str):
|
|
return {"expected": value}
|
|
elif isinstance(value, dict):
|
|
return value
|
|
elif isinstance(value, list):
|
|
return {"expected": value}
|
|
else:
|
|
raise ValueError("Invalid type")
|
|
|
|
|
|
def parse_checks(value):
|
|
# To avoid circular imports
|
|
from argos.checks import get_registered_checks
|
|
|
|
available_names = get_registered_checks().keys()
|
|
|
|
for name, expected in value.items():
|
|
if name not in available_names:
|
|
msg = f"Check should be one of f{available_names}. ({name} given)"
|
|
raise ValueError(msg)
|
|
return (name, expected)
|
|
|
|
|
|
class WebsitePath(BaseModel):
|
|
path: str
|
|
checks: List[
|
|
Annotated[
|
|
Tuple[str, str | dict | int],
|
|
BeforeValidator(parse_checks),
|
|
]
|
|
]
|
|
|
|
|
|
class Website(BaseModel):
|
|
domain: HttpUrl
|
|
frequency: Optional[int] = None
|
|
paths: List[WebsitePath]
|
|
|
|
@field_validator("frequency", mode="before")
|
|
def parse_frequency(cls, value):
|
|
if value:
|
|
return string_to_duration(value, "hours")
|
|
|
|
|
|
class Service(BaseModel):
|
|
secrets: List[str]
|
|
|
|
|
|
class Alert(BaseModel):
|
|
error: List[str]
|
|
warning: List[str]
|
|
alert: List[str]
|
|
|
|
|
|
class General(BaseModel):
|
|
frequency: int
|
|
alerts: Alert
|
|
|
|
@field_validator("frequency", mode="before")
|
|
def parse_frequency(cls, value):
|
|
return string_to_duration(value, "minutes")
|
|
|
|
|
|
class Config(BaseModel):
|
|
general: General
|
|
service: Service
|
|
ssl: SSL
|
|
websites: List[Website]
|