File: test_stations6.py

package info (click to toggle)
python-airos 0.6.4-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 704 kB
  • sloc: python: 2,967; sh: 19; makefile: 3
file content (119 lines) | stat: -rw-r--r-- 3,815 bytes parent folder | download
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
"""Ubiquiti AirOS tests."""

from http.cookies import SimpleCookie
import json
import os
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import aiofiles
import pytest

from airos.airos6 import AirOS6
from airos.data import AirOS6Data


async def _read_fixture(fixture: str = "NanoStation_M5_sta_v6.3.16") -> Any:
    """Read fixture file per device type."""
    fixture_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures", "userdata")
    path = os.path.join(fixture_dir, f"{fixture}.json")
    try:
        async with aiofiles.open(path, encoding="utf-8") as f:
            return json.loads(await f.read())
    except FileNotFoundError:
        pytest.fail(f"Fixture file not found: {path}")
    except json.JSONDecodeError as e:
        pytest.fail(f"Invalid JSON in fixture file {path}: {e}")


@pytest.mark.parametrize(
    ("mode", "fixture"),
    [
        ("sta", "NanoStation_loco_M5_v6.3.16_XM_sta"),
        ("ap", "NanoStation_loco_M5_v6.3.16_XM_ap"),
        ("sta", "NanoStation_M5_sta_v6.3.16"),
    ],
)
@pytest.mark.asyncio
async def test_ap_object(
    airos6_device: AirOS6, base_url: str, mode: str, fixture: str
) -> None:
    """Test device operation using the new _request_json method."""
    fixture_data = await _read_fixture(fixture)

    # Create an async mock that can return different values for different calls
    mock_request_json = AsyncMock(
        side_effect=[
            {},  # First call for login()
            fixture_data,  # Second call for status()
        ]
    )

    with (
        # Patch the internal method, not the session object
        patch.object(airos6_device, "_request_json", new=mock_request_json),
        # You need to manually set the connected state since login() is mocked
        patch.object(airos6_device, "connected", True),
    ):
        # We don't need to patch the session directly anymore
        await airos6_device.login()
        status: AirOS6Data = await airos6_device.status()

    # Assertions remain the same as they check the final result
    assert status.wireless.mode
    assert status.wireless.mode.value == mode
    assert status.derived.mac_interface == "br0"

    cookie = SimpleCookie()
    cookie["session_id"] = "test-cookie"
    cookie["AIROS_TOKEN"] = "abc123"


@pytest.mark.asyncio
async def test_login_v6_flow() -> None:
    """Test AirOS v6 XM login flow with manual cookie handling."""

    # Create a mock session
    session = MagicMock()

    # Mock response for GET /login.cgi
    get_login_response = MagicMock()
    get_login_response.__aenter__.return_value = get_login_response
    get_login_response.status = 200
    get_login_response.cookies = {
        "AIROS_ABC123": MagicMock(key="AIROS_ABC123", value="xyz789")
    }

    # Mock response for POST /login.cgi
    post_login_response = MagicMock()
    post_login_response.__aenter__.return_value = post_login_response
    post_login_response.status = 302

    # Mock response for GET /index.cgi
    get_index_response = MagicMock()
    get_index_response.__aenter__.return_value = get_index_response
    get_index_response.status = 200
    get_index_response.url = "http://192.168.1.3/index.cgi"

    # Set side effects for session.request
    session.request.side_effect = [
        get_login_response,
        post_login_response,
        get_index_response,
    ]

    # Create device
    airos6_device = AirOS6(
        host="http://192.168.1.3",
        username="ubnt",
        password="ubnt",
        session=session,
    )

    await airos6_device._login_v6()  # noqa: SLF001

    # Assertions
    assert airos6_device.connected is True
    assert airos6_device.api_version == 6
    assert airos6_device._auth_cookie == "AIROS_ABC123=xyz789"  # noqa: SLF001
    assert session.request.call_count == 3