Compare commits
3 Commits
a55fc938f0
...
f446ae52e7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f446ae52e7 | ||
|
|
c4f20d3241 | ||
|
|
bebbe78163 |
@@ -14,7 +14,7 @@ default_system = {
|
|||||||
</vars>
|
</vars>
|
||||||
|
|
||||||
<group name="system" default="default_system">
|
<group name="system" default="default_system">
|
||||||
# VRP (R) software, Version {{ version }} {{ _line_ }}
|
# VRP (R) software, Version {{ version }} ({{ model }} {{ _line_ }}
|
||||||
# ESN of slot {{ slot_number }}: {{ serial_number }}
|
# ESN of slot {{ slot_number }}: {{ serial_number }}
|
||||||
</group>
|
</group>
|
||||||
|
|
||||||
|
|||||||
25
oxi/interfaces/utils.py
Normal file
25
oxi/interfaces/utils.py
Normal file
@@ -0,0 +1,25 @@
|
|||||||
|
def expand_vlan_range(value: str | list[str]) -> list[str]:
|
||||||
|
"""Expand values like '1,7,14-15' into individual VLAN IDs."""
|
||||||
|
if isinstance(value, list):
|
||||||
|
value = ",".join(str(item) for item in value)
|
||||||
|
|
||||||
|
result: list[str] = []
|
||||||
|
if not value:
|
||||||
|
return result
|
||||||
|
for part in value.split(","):
|
||||||
|
part = part.strip()
|
||||||
|
if not part:
|
||||||
|
continue
|
||||||
|
if "-" in part:
|
||||||
|
start_s, end_s = part.split("-", 1)
|
||||||
|
try:
|
||||||
|
start, end = int(start_s), int(end_s)
|
||||||
|
except ValueError:
|
||||||
|
result.append(part)
|
||||||
|
continue
|
||||||
|
if start > end:
|
||||||
|
start, end = end, start
|
||||||
|
result.extend(str(i) for i in range(start, end + 1))
|
||||||
|
else:
|
||||||
|
result.append(part)
|
||||||
|
return result
|
||||||
87
tests/test_network.py
Normal file
87
tests/test_network.py
Normal file
@@ -0,0 +1,87 @@
|
|||||||
|
import pytest
|
||||||
|
import responses
|
||||||
|
|
||||||
|
from conftest import load
|
||||||
|
from oxi import OxiAPI
|
||||||
|
from oxi.exception import OxiAPIError
|
||||||
|
|
||||||
|
BASE = "https://oxi.example.com"
|
||||||
|
|
||||||
|
NODE_DATA = {
|
||||||
|
"name": "HQ",
|
||||||
|
"full_name": "grp/HQ",
|
||||||
|
"model": "keenetic",
|
||||||
|
"ip": "192.168.1.1",
|
||||||
|
"group": "grp",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_node_show_returns_view():
|
||||||
|
responses.get(f"{BASE}/node/show/HQ.json", json=NODE_DATA)
|
||||||
|
|
||||||
|
api = OxiAPI(url=BASE)
|
||||||
|
node = api.node("HQ")
|
||||||
|
|
||||||
|
assert node.ip == "192.168.1.1"
|
||||||
|
assert node.model == "keenetic"
|
||||||
|
assert node.full_name == "grp/HQ"
|
||||||
|
assert node.group == "grp"
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_node_config_fetches_and_parses():
|
||||||
|
responses.get(f"{BASE}/node/show/HQ.json", json=NODE_DATA)
|
||||||
|
responses.get(f"{BASE}/node/fetch/grp/HQ", body=load("keenetic"))
|
||||||
|
|
||||||
|
api = OxiAPI(url=BASE)
|
||||||
|
config = api.node("HQ").config
|
||||||
|
|
||||||
|
assert config.system.model == "Sprinter (KN-3710)"
|
||||||
|
assert len(config.interfaces) > 0
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_node_not_found_maps_to_404():
|
||||||
|
responses.get(f"{BASE}/node/show/missing.json", status=404)
|
||||||
|
|
||||||
|
api = OxiAPI(url=BASE)
|
||||||
|
with pytest.raises(OxiAPIError) as exc:
|
||||||
|
api.node("missing")
|
||||||
|
|
||||||
|
assert exc.value.status_code == 404
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_500_with_node_not_found_html_maps_to_404():
|
||||||
|
responses.get(
|
||||||
|
f"{BASE}/node/show/ghost.json",
|
||||||
|
status=500,
|
||||||
|
content_type="text/html",
|
||||||
|
body="<html><title>Oxidized::NodeNotFound</title></html>",
|
||||||
|
)
|
||||||
|
|
||||||
|
api = OxiAPI(url=BASE)
|
||||||
|
with pytest.raises(OxiAPIError) as exc:
|
||||||
|
api.node("ghost")
|
||||||
|
|
||||||
|
assert exc.value.status_code == 404
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_reload_returns_status_code():
|
||||||
|
responses.get(f"{BASE}/reload", status=200)
|
||||||
|
|
||||||
|
api = OxiAPI(url=BASE)
|
||||||
|
assert api.reload() == 200
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_unknown_model_raises_value_error():
|
||||||
|
data = {**NODE_DATA, "model": "unknown_vendor"}
|
||||||
|
responses.get(f"{BASE}/node/show/HQ.json", json=data)
|
||||||
|
responses.get(f"{BASE}/node/fetch/grp/HQ", body="whatever")
|
||||||
|
|
||||||
|
api = OxiAPI(url=BASE)
|
||||||
|
with pytest.raises(ValueError, match="not found in registry"):
|
||||||
|
api.node("HQ").config
|
||||||
71
tests/test_units.py
Normal file
71
tests/test_units.py
Normal file
@@ -0,0 +1,71 @@
|
|||||||
|
import pytest
|
||||||
|
|
||||||
|
from conftest import load
|
||||||
|
from oxi.exception import OxiAPIError
|
||||||
|
from oxi.interfaces import device_registry
|
||||||
|
from oxi.interfaces.base import BaseDevice
|
||||||
|
from oxi.interfaces.contract import Interfaces, System
|
||||||
|
from oxi.interfaces.utils import expand_vlan_range as eltex_expand
|
||||||
|
from oxi.interfaces.utils import expand_vlan_range as qtech_expand
|
||||||
|
|
||||||
|
|
||||||
|
class TestExpandVlanRange:
|
||||||
|
@pytest.mark.parametrize("expand", [qtech_expand, eltex_expand])
|
||||||
|
def test_simple_and_range(self, expand):
|
||||||
|
assert expand("1,7,14-15") == ["1", "7", "14", "15"]
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("expand", [qtech_expand, eltex_expand])
|
||||||
|
def test_reversed_range_is_normalized(self, expand):
|
||||||
|
assert expand("15-13") == ["13", "14", "15"]
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("expand", [qtech_expand, eltex_expand])
|
||||||
|
def test_non_numeric_range_kept_verbatim(self, expand):
|
||||||
|
assert expand("a-b") == ["a-b"]
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("expand", [qtech_expand, eltex_expand])
|
||||||
|
def test_empty(self, expand):
|
||||||
|
assert expand("") == []
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("expand", [qtech_expand, eltex_expand])
|
||||||
|
def test_list_input(self, expand):
|
||||||
|
assert expand(["1", "3-4"]) == ["1", "3", "4"]
|
||||||
|
|
||||||
|
|
||||||
|
class TestKeeneticDecodeUtf:
|
||||||
|
@pytest.fixture(scope="class")
|
||||||
|
def keenetic(self):
|
||||||
|
return device_registry["keenetic"](load("keenetic"))
|
||||||
|
|
||||||
|
def test_plain_text_passthrough(self, keenetic):
|
||||||
|
assert keenetic._decode_utf("Plain ASCII") == "Plain ASCII"
|
||||||
|
|
||||||
|
def test_escaped_utf8_is_decoded(self, keenetic):
|
||||||
|
assert keenetic._decode_utf(r'"\xd0\x94\xd0\xbe\xd0\xbc"') == "Дом"
|
||||||
|
|
||||||
|
|
||||||
|
class TestTemplateValidation:
|
||||||
|
def test_missing_required_group_raises(self):
|
||||||
|
class OnlySystem(BaseDevice):
|
||||||
|
template = "dummy.ttp"
|
||||||
|
|
||||||
|
def _load_template(self):
|
||||||
|
return '<group name="system"></group>'
|
||||||
|
|
||||||
|
with pytest.raises(ValueError, match="missing required groups"):
|
||||||
|
OnlySystem("data")
|
||||||
|
|
||||||
|
def test_missing_template_file_raises(self):
|
||||||
|
class NoTemplate(BaseDevice):
|
||||||
|
template = "does_not_exist.ttp"
|
||||||
|
|
||||||
|
with pytest.raises(FileNotFoundError):
|
||||||
|
NoTemplate("data")
|
||||||
|
|
||||||
|
|
||||||
|
class TestNodeNotFound:
|
||||||
|
def test_not_found_config_raises_on_parse(self):
|
||||||
|
device = device_registry["eltex"](load("eltex", "not_found.conf"), name="HQ")
|
||||||
|
assert device.raw is None
|
||||||
|
with pytest.raises(OxiAPIError) as exc:
|
||||||
|
device.parse()
|
||||||
|
assert exc.value.status_code == 404
|
||||||
Reference in New Issue
Block a user