SyllasGS/V-Fire-Monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add production installer and observability tooling

Browse Source
This commit is contained in:
SyllasGS
2026-05-07 12:11:15 -03:00
parent a19ac9f00c
commit 5d92a6c99a
30 changed files with 3493 additions and 1015 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_core.py
+137 -53
View File
@@ -3,90 +3,127 @@ from pathlib import Path
import pytest
from cryptography.fernet import Fernet
from werkzeug.security import check_password_hash
from vfire_monitor import create_app
from werkzeug.security import check_password_hash
from vfire_monitor import create_app
from vfire_monitor.core import (
LicenseService,
contiguous_ranges,
load_license_key,
normalize_config,
validate_login_payload,
validate_node_payload,
)
def test_validate_node_payload_normalizes_fields():
node = validate_node_payload({"nome": " Central A ", "ip": " 10.0.0.1 ", "unit": "3"})
assert node == {"nome": "Central A", "ip": "10.0.0.1", "unit": 3}
assert node == {"nome": "Central A", "ip": "10.0.0.1", "unit": 3, "port": 502}
def test_validate_node_payload_rejects_invalid_unit():
with pytest.raises(ValueError):
validate_node_payload({"nome": "Central A", "ip": "10.0.0.1", "unit": 999})
def test_validate_node_payload_rejects_invalid_ip():
with pytest.raises(ValueError):
validate_node_payload({"nome": "Central A", "ip": "999.1.1.1", "unit": 3})
def test_validate_node_payload_accepts_custom_port():
node = validate_node_payload({"nome": "Central A", "ip": "10.0.0.1", "unit": 3, "port": "1502"})
assert node["port"] == 1502
def test_normalize_config_migrates_legacy_password():
config = normalize_config(
{
config = normalize_config(
{
"web_user": "admin",
"web_password": "segredo",
"nodes": [{"nome": "Node 1", "ip": "1.1.1.1", "unit": 1}],
},
default_password="admin",
)
assert "web_password" not in config
assert check_password_hash(config["web_password_hash"], "segredo")
assert config["nodes"][0]["nome"] == "Node 1"
def test_contiguous_ranges_splits_large_blocks():
ranges = contiguous_ranges(list(range(1, 125)))
assert ranges == [(1, 120), (121, 124)]
assert "web_password" not in config
assert check_password_hash(config["web_password_hash"], "segredo")
assert config["nodes"][0]["nome"] == "Node 1"
def test_contiguous_ranges_splits_large_blocks():
ranges = contiguous_ranges(list(range(1, 125)))
assert ranges == [(1, 120), (121, 124)]
def test_license_service_validates_serial(tmp_path, monkeypatch):
monkeypatch.setattr("vfire_monitor.core.get_hardware_id", lambda: "HW-123")
key = Fernet.generate_key()
payload = {"h": "HW-123", "c": "Cliente", "e": "2099-01-01"}
serial = Fernet(key).encrypt(json.dumps(payload).encode("utf-8")).decode("utf-8")
license_path = tmp_path / "license.key"
license_path.write_text(serial, encoding="utf-8")
service = LicenseService(license_path, key)
status = service.verify()
monkeypatch.setattr("vfire_monitor.core.get_hardware_id", lambda: "HW-123")
key = Fernet.generate_key()
payload = {"h": "HW-123", "c": "Cliente", "e": "2099-01-01"}
serial = Fernet(key).encrypt(json.dumps(payload).encode("utf-8")).decode("utf-8")
license_path = tmp_path / "license.key"
license_path.write_text(serial, encoding="utf-8")
service = LicenseService(license_path, key)
status = service.verify()
assert status.valid is True
assert "Cliente" in status.message
def test_license_service_install_rejects_wrong_hwid(tmp_path, monkeypatch):
monkeypatch.setattr("vfire_monitor.core.get_hardware_id", lambda: "HW-123")
key = Fernet.generate_key()
payload = {"h": "HW-999", "c": "Cliente", "e": "2099-01-01"}
serial = Fernet(key).encrypt(json.dumps(payload).encode("utf-8")).decode("utf-8")
service = LicenseService(tmp_path / "license.key", key)
with pytest.raises(ValueError):
service.install(serial)
def test_load_license_key_requires_env_in_production(monkeypatch):
monkeypatch.delenv("VFM_LICENSE_MASTER_KEY", raising=False)
with pytest.raises(RuntimeError):
load_license_key(require_env=True)
def test_validate_login_payload_requires_minimum_password_length():
with pytest.raises(ValueError):
validate_login_payload("admin", "1234567")
@pytest.fixture
def app(tmp_path, monkeypatch):
monkeypatch.setenv("VFM_DEFAULT_PASSWORD", "admin")
monkeypatch.setenv("VFM_LICENSE_MASTER_KEY", Fernet.generate_key().decode("utf-8"))
application = create_app(base_dir=tmp_path, start_monitor=False)
application.config["TESTING"] = True
runtime_state = application.extensions["runtime_state"]
runtime_state.config["web_user"] = "admin"
runtime_state.config["web_password_hash"] = normalize_config({}, "admin")["web_password_hash"]
runtime_state.save_config()
return application
@pytest.fixture
monkeypatch.setenv("VFM_DEFAULT_PASSWORD", "admin")
monkeypatch.setenv("VFM_LICENSE_MASTER_KEY", Fernet.generate_key().decode("utf-8"))
application = create_app(base_dir=tmp_path, start_monitor=False)
application.config["TESTING"] = True
runtime_state = application.extensions["runtime_state"]
runtime_state.config["web_user"] = "admin"
runtime_state.config["web_password_hash"] = normalize_config({}, "admin")["web_password_hash"]
runtime_state.save_config()
return application
@pytest.fixture
def client(app):
return app.test_client()
def csrf_token(client, path="/login"):
client.get(path)
with client.session_transaction() as session:
return session["csrf_token"]
def login(client):
return client.post("/login", data={"username": "admin", "password": "admin"}, follow_redirects=False)
return client.post(
"/login",
data={"username": "admin", "password": "admin", "csrf_token": csrf_token(client)},
follow_redirects=False,
)
def test_login_success(client):
@@ -94,26 +131,73 @@ def test_login_success(client):
assert response.status_code == 302
def test_login_requires_csrf(client):
response = client.post("/login", data={"username": "admin", "password": "admin"}, follow_redirects=False)
assert response.status_code == 302
def test_api_nodes_requires_auth(client):
response = client.post("/api/nodes", json={})
response = client.post("/api/nodes", json={}, headers={"X-CSRF-Token": csrf_token(client)})
assert response.status_code == 302
def test_api_nodes_updates_configuration(client, app):
login(client)
response = client.post(
"/api/nodes",
response = client.post(
"/api/nodes",
json={
"zabbix_server": "127.0.0.1",
"hostname_zabbix": "HOST-01",
"nodes": [{"nome": "Central A", "ip": "10.0.0.5", "unit": 3}],
"nodes": [{"nome": "Central A", "ip": "10.0.0.5", "unit": 3, "port": 1502}],
},
headers={"X-CSRF-Token": csrf_token(client, "/")},
)
assert response.status_code == 200
assert response.get_json()["status"] == "success"
config_path = Path(app.extensions["runtime_state"].config_store.path)
config = json.loads(config_path.read_text(encoding="utf-8"))
assert config["hostname_zabbix"] == "HOST-01"
assert config["nodes"][0]["nome"] == "Central A"
assert config["nodes"][0]["port"] == 1502
def test_api_nodes_rejects_missing_csrf(client):
login(client)
response = client.post("/api/nodes", json={"zabbix_server": "127.0.0.1", "hostname_zabbix": "HOST-01", "nodes": []})
assert response.status_code == 403
def test_healthz_is_public(client):
response = client.get("/healthz")
assert response.status_code == 200
assert response.get_json()["status"] == "ok"
def test_api_account_updates_credentials(client, app):
login(client)
response = client.post(
"/api/account",
json={
"username": "operador",
"current_password": "admin",
"new_password": "senha-forte-123",
},
headers={"X-CSRF-Token": csrf_token(client, "/")},
)
assert response.status_code == 200
assert response.get_json()["status"] == "success"
config_path = Path(app.extensions["runtime_state"].config_store.path)
config = json.loads(config_path.read_text(encoding="utf-8"))
assert config["hostname_zabbix"] == "HOST-01"
assert config["nodes"][0]["nome"] == "Central A"
runtime_state = app.extensions["runtime_state"]
assert runtime_state.authenticate("operador", "senha-forte-123") is True
def test_api_system_status_returns_monitor_snapshot(client):
login(client)
response = client.get("/api/system/status")
assert response.status_code == 200
body = response.get_json()
assert body["status"] == "success"
assert "monitor" in body