Skip to content
Snippets Groups Projects
Select Git revision
  • develop default
  • main protected
2 results
Blame Permalink
test_interfaces_routes.py 3.01 KiB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95 






import jsonschema

from resource_management.routes.default import Version
from resource_management.routes import interfaces
from resource_management import router_interfaces
from resource_management import db
from resource_management.db import model

import pytest


def test_bad_method(client):
    rv = client.post('/api/version')
    assert rv.status_code == 405


def test_version_request(client):
    rv = client.get('/api/version')
    assert rv.status_code == 200
    jsonschema.validate(rv.json(), Version.schema())


def test_update_router_interfaces(
        client, resources_db, mocked_router, router_name):
    rv = client.post(f'/api/interfaces/initialize-router/{router_name}')
    assert rv.status_code == 200
    jsonschema.validate(rv.json(), interfaces.InterfacesSummary.schema())


def test_next_lag(client, resources_db, mocked_router, router_name):

    router_interfaces.load_new_router_interfaces(router_name)

    rv = client.post(f'/api/interfaces/next-lag/{router_name}')
    assert rv.status_code == 200
    response = rv.json()
    jsonschema.validate(response, interfaces.NextLAG.schema())

    with db.session_scope() as session:
        row = session.query(model.LAG). \
            filter_by(name=response['name']). \
            join(model.Router). \
            filter_by(fqdn=router_name).one()
        assert row.availability == model.AvalabilityStates.RESERVED.name


def test_next_physical(client, resources_db, mocked_router, router_name):

    if '.lab.' in router_name:
        pytest.skip('not all lab routers have available ports')
    if router_name.startswith('rt'):
        pytest.skip('not all rt* have available ports')

    router_interfaces.load_new_router_interfaces(router_name)

    with db.session_scope() as session:
        rows = session.query(model.LAG.name). \
            join(model.Router). \
            filter_by(fqdn=router_name).all()
        lag_name = rows[0][0]

    rv = client.post(f'/api/interfaces/next-physical/{router_name}/{lag_name}')
    assert rv.status_code == 200

    response = rv.json()
    jsonschema.validate(response, interfaces.NextPhysicalInterface.schema())

    assert response['fqdn'] == router_name
    assert response['lag'] == lag_name

    with db.session_scope() as session:
        ifc_row = session.query(model.PhysicalInterface). \
            filter_by(name=response['name']). \
            join(model.Router). \
            filter_by(fqdn=router_name). \
            join(model.LAG). \
            filter_by(name=lag_name).one()
        assert ifc_row.availability == model.AvalabilityStates.RESERVED.name


def test_list_of_routers(client, resources_db):

    with db.session_scope() as session:
        for idx in range(10):
            row = db.model.Router(fqdn=f'router-{idx}.xyz.com')
            session.add(row)

    rv = client.get('/api/interfaces/routers')
    assert rv.status_code == 200

    response = rv.json()
    jsonschema.validate(response, interfaces.ListOfRouters.schema())
    for r in response['routers']:
        assert r['fqdn'].endswith('.xyz.com')