582 lines
18 KiB
Python
582 lines
18 KiB
Python
from os import environ
|
|
from pathlib import Path
|
|
from typing import List, Union
|
|
from unittest import mock
|
|
|
|
from pydantic import BaseModel, Field
|
|
from pytest import mark, raises
|
|
from typing_extensions import Annotated
|
|
|
|
from libretime_shared.config import (
|
|
AnyHttpUrlStr,
|
|
BaseConfig,
|
|
DatabaseConfig,
|
|
IcecastOutput,
|
|
RabbitMQConfig,
|
|
ShoutcastOutput,
|
|
)
|
|
|
|
AnyOutput = Annotated[
|
|
Union[IcecastOutput, ShoutcastOutput],
|
|
Field(discriminator="kind"),
|
|
]
|
|
|
|
|
|
# pylint: disable=too-few-public-methods
|
|
class FixtureConfig(BaseConfig):
|
|
public_url: AnyHttpUrlStr
|
|
api_key: str
|
|
allowed_hosts: List[str] = []
|
|
database: DatabaseConfig
|
|
rabbitmq: RabbitMQConfig = RabbitMQConfig()
|
|
outputs: List[AnyOutput]
|
|
|
|
|
|
FIXTURE_CONFIG_JSON_SCHEMA = {
|
|
"$defs": {
|
|
"AudioAAC": {
|
|
"properties": {
|
|
"channels": {
|
|
"allOf": [{"$ref": "#/$defs/AudioChannels"}],
|
|
"default": "stereo",
|
|
},
|
|
"bitrate": {"title": "Bitrate", "type": "integer"},
|
|
"format": {
|
|
"const": "aac",
|
|
"default": "aac",
|
|
"enum": ["aac"],
|
|
"title": "Format",
|
|
"type": "string",
|
|
},
|
|
},
|
|
"required": ["bitrate"],
|
|
"title": "AudioAAC",
|
|
"type": "object",
|
|
},
|
|
"AudioChannels": {
|
|
"enum": ["stereo", "mono"],
|
|
"title": "AudioChannels",
|
|
"type": "string",
|
|
},
|
|
"AudioMP3": {
|
|
"properties": {
|
|
"channels": {
|
|
"allOf": [{"$ref": "#/$defs/AudioChannels"}],
|
|
"default": "stereo",
|
|
},
|
|
"bitrate": {"title": "Bitrate", "type": "integer"},
|
|
"format": {
|
|
"const": "mp3",
|
|
"default": "mp3",
|
|
"enum": ["mp3"],
|
|
"title": "Format",
|
|
"type": "string",
|
|
},
|
|
},
|
|
"required": ["bitrate"],
|
|
"title": "AudioMP3",
|
|
"type": "object",
|
|
},
|
|
"AudioOGG": {
|
|
"properties": {
|
|
"channels": {
|
|
"allOf": [{"$ref": "#/$defs/AudioChannels"}],
|
|
"default": "stereo",
|
|
},
|
|
"bitrate": {"title": "Bitrate", "type": "integer"},
|
|
"format": {
|
|
"const": "ogg",
|
|
"default": "ogg",
|
|
"enum": ["ogg"],
|
|
"title": "Format",
|
|
"type": "string",
|
|
},
|
|
"enable_metadata": {
|
|
"anyOf": [{"type": "boolean"}, {"type": "null"}],
|
|
"default": False,
|
|
"title": "Enable Metadata",
|
|
},
|
|
},
|
|
"required": ["bitrate"],
|
|
"title": "AudioOGG",
|
|
"type": "object",
|
|
},
|
|
"AudioOpus": {
|
|
"properties": {
|
|
"channels": {
|
|
"allOf": [{"$ref": "#/$defs/AudioChannels"}],
|
|
"default": "stereo",
|
|
},
|
|
"bitrate": {"title": "Bitrate", "type": "integer"},
|
|
"format": {
|
|
"const": "opus",
|
|
"default": "opus",
|
|
"enum": ["opus"],
|
|
"title": "Format",
|
|
"type": "string",
|
|
},
|
|
},
|
|
"required": ["bitrate"],
|
|
"title": "AudioOpus",
|
|
"type": "object",
|
|
},
|
|
"DatabaseConfig": {
|
|
"properties": {
|
|
"host": {
|
|
"default": "localhost",
|
|
"title": "Host",
|
|
"type": "string",
|
|
},
|
|
"port": {"default": 5432, "title": "Port", "type": "integer"},
|
|
"name": {
|
|
"default": "libretime",
|
|
"title": "Name",
|
|
"type": "string",
|
|
},
|
|
"user": {
|
|
"default": "libretime",
|
|
"title": "User",
|
|
"type": "string",
|
|
},
|
|
"password": {
|
|
"default": "libretime",
|
|
"title": "Password",
|
|
"type": "string",
|
|
},
|
|
},
|
|
"title": "DatabaseConfig",
|
|
"type": "object",
|
|
},
|
|
"IcecastOutput": {
|
|
"properties": {
|
|
"kind": {
|
|
"const": "icecast",
|
|
"default": "icecast",
|
|
"enum": ["icecast"],
|
|
"title": "Kind",
|
|
"type": "string",
|
|
},
|
|
"enabled": {
|
|
"default": False,
|
|
"title": "Enabled",
|
|
"type": "boolean",
|
|
},
|
|
"public_url": {
|
|
"anyOf": [
|
|
{"format": "uri", "type": "string"},
|
|
{"type": "null"},
|
|
],
|
|
"default": None,
|
|
"title": "Public Url",
|
|
},
|
|
"host": {
|
|
"default": "localhost",
|
|
"title": "Host",
|
|
"type": "string",
|
|
},
|
|
"port": {"default": 8000, "title": "Port", "type": "integer"},
|
|
"mount": {"title": "Mount", "type": "string"},
|
|
"source_user": {
|
|
"default": "source",
|
|
"title": "Source User",
|
|
"type": "string",
|
|
},
|
|
"source_password": {
|
|
"title": "Source Password",
|
|
"type": "string",
|
|
},
|
|
"admin_user": {
|
|
"default": "admin",
|
|
"title": "Admin User",
|
|
"type": "string",
|
|
},
|
|
"admin_password": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"default": None,
|
|
"title": "Admin Password",
|
|
},
|
|
"audio": {
|
|
"discriminator": {
|
|
"mapping": {
|
|
"aac": "#/$defs/AudioAAC",
|
|
"mp3": "#/$defs/AudioMP3",
|
|
"ogg": "#/$defs/AudioOGG",
|
|
"opus": "#/$defs/AudioOpus",
|
|
},
|
|
"propertyName": "format",
|
|
},
|
|
"oneOf": [
|
|
{"$ref": "#/$defs/AudioAAC"},
|
|
{"$ref": "#/$defs/AudioMP3"},
|
|
{"$ref": "#/$defs/AudioOGG"},
|
|
{"$ref": "#/$defs/AudioOpus"},
|
|
],
|
|
"title": "Audio",
|
|
},
|
|
"name": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"default": None,
|
|
"title": "Name",
|
|
},
|
|
"description": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"default": None,
|
|
"title": "Description",
|
|
},
|
|
"website": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"default": None,
|
|
"title": "Website",
|
|
},
|
|
"genre": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"default": None,
|
|
"title": "Genre",
|
|
},
|
|
"mobile": {
|
|
"default": False,
|
|
"title": "Mobile",
|
|
"type": "boolean",
|
|
},
|
|
},
|
|
"required": ["mount", "source_password", "audio"],
|
|
"title": "IcecastOutput",
|
|
"type": "object",
|
|
},
|
|
"RabbitMQConfig": {
|
|
"properties": {
|
|
"host": {
|
|
"default": "localhost",
|
|
"title": "Host",
|
|
"type": "string",
|
|
},
|
|
"port": {"default": 5672, "title": "Port", "type": "integer"},
|
|
"user": {
|
|
"default": "libretime",
|
|
"title": "User",
|
|
"type": "string",
|
|
},
|
|
"password": {
|
|
"default": "libretime",
|
|
"title": "Password",
|
|
"type": "string",
|
|
},
|
|
"vhost": {
|
|
"default": "/libretime",
|
|
"title": "Vhost",
|
|
"type": "string",
|
|
},
|
|
},
|
|
"title": "RabbitMQConfig",
|
|
"type": "object",
|
|
},
|
|
"ShoutcastOutput": {
|
|
"properties": {
|
|
"kind": {
|
|
"const": "shoutcast",
|
|
"default": "shoutcast",
|
|
"enum": ["shoutcast"],
|
|
"title": "Kind",
|
|
"type": "string",
|
|
},
|
|
"enabled": {
|
|
"default": False,
|
|
"title": "Enabled",
|
|
"type": "boolean",
|
|
},
|
|
"public_url": {
|
|
"anyOf": [
|
|
{"format": "uri", "type": "string"},
|
|
{"type": "null"},
|
|
],
|
|
"default": None,
|
|
"title": "Public Url",
|
|
},
|
|
"host": {
|
|
"default": "localhost",
|
|
"title": "Host",
|
|
"type": "string",
|
|
},
|
|
"port": {"default": 8000, "title": "Port", "type": "integer"},
|
|
"source_user": {
|
|
"default": "source",
|
|
"title": "Source User",
|
|
"type": "string",
|
|
},
|
|
"source_password": {
|
|
"title": "Source Password",
|
|
"type": "string",
|
|
},
|
|
"admin_user": {
|
|
"default": "admin",
|
|
"title": "Admin User",
|
|
"type": "string",
|
|
},
|
|
"admin_password": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"default": None,
|
|
"title": "Admin Password",
|
|
},
|
|
"audio": {
|
|
"discriminator": {
|
|
"mapping": {
|
|
"aac": "#/$defs/AudioAAC",
|
|
"mp3": "#/$defs/AudioMP3",
|
|
},
|
|
"propertyName": "format",
|
|
},
|
|
"oneOf": [
|
|
{"$ref": "#/$defs/AudioAAC"},
|
|
{"$ref": "#/$defs/AudioMP3"},
|
|
],
|
|
"title": "Audio",
|
|
},
|
|
"name": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"default": None,
|
|
"title": "Name",
|
|
},
|
|
"website": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"default": None,
|
|
"title": "Website",
|
|
},
|
|
"genre": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"default": None,
|
|
"title": "Genre",
|
|
},
|
|
"mobile": {
|
|
"default": False,
|
|
"title": "Mobile",
|
|
"type": "boolean",
|
|
},
|
|
},
|
|
"required": ["source_password", "audio"],
|
|
"title": "ShoutcastOutput",
|
|
"type": "object",
|
|
},
|
|
},
|
|
"properties": {
|
|
"public_url": {"format": "uri", "title": "Public Url", "type": "string"},
|
|
"api_key": {"title": "Api Key", "type": "string"},
|
|
"allowed_hosts": {
|
|
"default": [],
|
|
"items": {"type": "string"},
|
|
"title": "Allowed Hosts",
|
|
"type": "array",
|
|
},
|
|
"database": {"$ref": "#/$defs/DatabaseConfig"},
|
|
"rabbitmq": {
|
|
"allOf": [{"$ref": "#/$defs/RabbitMQConfig"}],
|
|
"default": {
|
|
"host": "localhost",
|
|
"port": 5672,
|
|
"user": "libretime",
|
|
"password": "libretime",
|
|
"vhost": "/libretime",
|
|
},
|
|
},
|
|
"outputs": {
|
|
"items": {
|
|
"discriminator": {
|
|
"mapping": {
|
|
"icecast": "#/$defs/IcecastOutput",
|
|
"shoutcast": "#/$defs/ShoutcastOutput",
|
|
},
|
|
"propertyName": "kind",
|
|
},
|
|
"oneOf": [
|
|
{"$ref": "#/$defs/IcecastOutput"},
|
|
{"$ref": "#/$defs/ShoutcastOutput"},
|
|
],
|
|
},
|
|
"title": "Outputs",
|
|
"type": "array",
|
|
},
|
|
},
|
|
"required": ["public_url", "api_key", "database", "outputs"],
|
|
"title": "FixtureConfig",
|
|
"type": "object",
|
|
}
|
|
|
|
|
|
FIXTURE_CONFIG_RAW = """
|
|
public_url: http://libretime.example.org/
|
|
api_key: "f3bf04fc"
|
|
allowed_hosts:
|
|
- example.com
|
|
- sub.example.com
|
|
|
|
# Comment !
|
|
database:
|
|
host: "localhost"
|
|
port: 5432
|
|
|
|
ignored: "ignored"
|
|
|
|
outputs:
|
|
- enabled: true
|
|
kind: icecast
|
|
host: localhost
|
|
port: 8000
|
|
mount: main.ogg
|
|
source_password: hackme
|
|
audio:
|
|
format: ogg
|
|
bitrate: 256
|
|
"""
|
|
|
|
|
|
def test_base_config(tmp_path: Path):
|
|
config_filepath = tmp_path / "config.yml"
|
|
config_filepath.write_text(FIXTURE_CONFIG_RAW)
|
|
|
|
with mock.patch.dict(
|
|
environ,
|
|
{
|
|
"LIBRETIME_API": "invalid",
|
|
"LIBRETIME_DATABASE_PORT": "8888",
|
|
"LIBRETIME_DATABASE": "invalid",
|
|
"LIBRETIME_RABBITMQ": "invalid",
|
|
"LIBRETIME_RABBITMQ_HOST": "changed",
|
|
"LIBRETIME_OUTPUTS_0_ENABLED": "false",
|
|
"LIBRETIME_OUTPUTS_0_HOST": "changed",
|
|
"WRONGPREFIX_API_KEY": "invalid",
|
|
},
|
|
):
|
|
config = FixtureConfig(config_filepath)
|
|
|
|
assert config.model_json_schema() == FIXTURE_CONFIG_JSON_SCHEMA
|
|
|
|
assert config.public_url == "http://libretime.example.org"
|
|
assert config.api_key == "f3bf04fc"
|
|
assert config.allowed_hosts == ["example.com", "sub.example.com"]
|
|
assert config.database.host == "localhost"
|
|
assert config.database.port == 8888
|
|
assert config.rabbitmq.host == "changed"
|
|
assert config.rabbitmq.port == 5672
|
|
assert config.outputs[0].enabled is False
|
|
assert config.outputs[0].kind == "icecast"
|
|
assert config.outputs[0].host == "changed"
|
|
assert config.outputs[0].audio.format == "ogg"
|
|
|
|
# Optional model: loading default values (rabbitmq)
|
|
with mock.patch.dict(environ, {}):
|
|
config = FixtureConfig(config_filepath)
|
|
assert config.allowed_hosts == ["example.com", "sub.example.com"]
|
|
assert config.rabbitmq.host == "localhost"
|
|
assert config.rabbitmq.port == 5672
|
|
|
|
# Optional model: overriding using environment (rabbitmq)
|
|
with mock.patch.dict(
|
|
environ,
|
|
{
|
|
"LIBRETIME_RABBITMQ_HOST": "changed",
|
|
"LIBRETIME_ALLOWED_HOSTS": "example.com, changed.example.com",
|
|
},
|
|
):
|
|
config = FixtureConfig(config_filepath)
|
|
assert config.allowed_hosts == ["example.com", "changed.example.com"]
|
|
assert config.rabbitmq.host == "changed"
|
|
assert config.rabbitmq.port == 5672
|
|
|
|
|
|
# pylint: disable=too-few-public-methods
|
|
class RequiredModel(BaseModel):
|
|
api_key: str
|
|
with_default: str = "original"
|
|
|
|
|
|
# pylint: disable=too-few-public-methods
|
|
class FixtureWithRequiredSubmodelConfig(BaseConfig):
|
|
required: RequiredModel
|
|
|
|
|
|
FIXTURE_WITH_REQUIRED_SUBMODEL_CONFIG_RAW = """
|
|
required:
|
|
api_key: "test_key"
|
|
"""
|
|
|
|
|
|
def test_base_config_required_submodel(tmp_path: Path):
|
|
config_filepath = tmp_path / "config.yml"
|
|
config_filepath.write_text(FIXTURE_WITH_REQUIRED_SUBMODEL_CONFIG_RAW)
|
|
|
|
# With config file
|
|
with mock.patch.dict(environ, {}):
|
|
config = FixtureWithRequiredSubmodelConfig(config_filepath)
|
|
assert config.required.api_key == "test_key"
|
|
assert config.required.with_default == "original"
|
|
|
|
# With env variables
|
|
with mock.patch.dict(environ, {"LIBRETIME_REQUIRED_API_KEY": "test_key"}):
|
|
config = FixtureWithRequiredSubmodelConfig(None)
|
|
assert config.required.api_key == "test_key"
|
|
assert config.required.with_default == "original"
|
|
|
|
# With env variables override
|
|
with mock.patch.dict(environ, {"LIBRETIME_REQUIRED_API_KEY": "changed"}):
|
|
config = FixtureWithRequiredSubmodelConfig(config_filepath)
|
|
assert config.required.api_key == "changed"
|
|
assert config.required.with_default == "original"
|
|
|
|
# With env variables default override
|
|
with mock.patch.dict(
|
|
environ,
|
|
{
|
|
"LIBRETIME_REQUIRED_API_KEY": "changed",
|
|
"LIBRETIME_REQUIRED_WITH_DEFAULT": "changed",
|
|
},
|
|
):
|
|
config = FixtureWithRequiredSubmodelConfig(config_filepath)
|
|
assert config.required.api_key == "changed"
|
|
assert config.required.with_default == "changed"
|
|
|
|
# Raise validation error
|
|
with mock.patch.dict(environ, {}):
|
|
with raises(SystemExit):
|
|
FixtureWithRequiredSubmodelConfig(None)
|
|
|
|
|
|
def test_base_config_from_init() -> None:
|
|
class FromInitFixtureConfig(BaseConfig):
|
|
found: str
|
|
override: str
|
|
|
|
with mock.patch.dict(environ, {"LIBRETIME_OVERRIDE": "changed"}):
|
|
config = FromInitFixtureConfig(
|
|
found="changed",
|
|
override="invalid",
|
|
)
|
|
|
|
assert config.found == "changed"
|
|
assert config.override == "changed"
|
|
|
|
|
|
FIXTURE_CONFIG_RAW_MISSING = """
|
|
database:
|
|
host: "localhost"
|
|
"""
|
|
|
|
FIXTURE_CONFIG_RAW_INVALID = """
|
|
database
|
|
host: "localhost"
|
|
"""
|
|
|
|
|
|
@mark.parametrize(
|
|
"raw,exception",
|
|
[
|
|
(FIXTURE_CONFIG_RAW_INVALID, SystemExit),
|
|
(FIXTURE_CONFIG_RAW_MISSING, SystemExit),
|
|
],
|
|
)
|
|
def test_load_config_error(tmp_path: Path, raw, exception):
|
|
config_filepath = tmp_path / "config.yml"
|
|
config_filepath.write_text(raw)
|
|
|
|
with raises(exception):
|
|
with mock.patch.dict(environ, {}):
|
|
FixtureConfig(config_filepath)
|