diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..4b1c095 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,74 @@ +name: CI + +on: + push: + branches: [ main, develop ] + pull_request: + branches: [ main, develop ] + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ['3.8', '3.9', '3.10', '3.11', '3.12'] + + steps: + - uses: actions/checkout@v3 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v4 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e ".[dev]" + + - name: Lint with flake8 + run: | + # Stop the build if there are Python syntax errors or undefined names + flake8 src/ --count --select=E9,F63,F7,F82 --show-source --statistics + # Exit-zero treats all errors as warnings + flake8 src/ --count --exit-zero --max-complexity=10 --max-line-length=88 --statistics + + - name: Type check with mypy + run: | + mypy src/ + continue-on-error: true + + - name: Test with pytest + run: | + pytest --cov=unifi_client --cov-report=xml --cov-report=term + + - name: Upload coverage to Codecov + uses: codecov/codecov-action@v3 + with: + file: ./coverage.xml + flags: unittests + name: codecov-umbrella + fail_ci_if_error: false + + build: + needs: test + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v3 + + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: '3.11' + + - name: Install build tools + run: | + python -m pip install --upgrade pip + pip install build twine + + - name: Build package + run: python -m build + + - name: Check package + run: twine check dist/* diff --git a/.gitignore b/.gitignore index b7faf40..86f40c6 100644 --- a/.gitignore +++ b/.gitignore @@ -205,3 +205,5 @@ cython_debug/ marimo/_static/ marimo/_lsp/ __marimo__/ + +.DS_Store \ No newline at end of file diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 0000000..04aec97 --- /dev/null +++ b/MANIFEST.in @@ -0,0 +1,8 @@ +include README.md +include LICENSE +include pyproject.toml +recursive-include src *.py +recursive-include tests *.py +global-exclude __pycache__ +global-exclude *.py[co] +global-exclude .DS_Store diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..62a314a --- /dev/null +++ b/Makefile @@ -0,0 +1,61 @@ +.PHONY: help install install-dev test coverage lint format type-check clean build publish-test publish + +help: + @echo "Available commands:" + @echo " install - Install package" + @echo " install-dev - Install package with dev dependencies" + @echo " test - Run tests" + @echo " coverage - Run tests with coverage report" + @echo " lint - Run flake8 linter" + @echo " format - Format code with black" + @echo " type-check - Run mypy type checker" + @echo " clean - Remove build artifacts" + @echo " build - Build package" + @echo " publish-test - Publish to Test PyPI" + @echo " publish - Publish to PyPI" + +install: + pip install -e . + +install-dev: + pip install -e ".[dev]" + +test: + pytest -v + +coverage: + pytest --cov=unifi_client --cov-report=html --cov-report=term-missing + +lint: + flake8 src/ tests/ + +format: + black src/ tests/ + +format-check: + black --check src/ tests/ + +type-check: + mypy src/ + +clean: + rm -rf build/ + rm -rf dist/ + rm -rf *.egg-info + rm -rf .pytest_cache/ + rm -rf .mypy_cache/ + rm -rf htmlcov/ + rm -rf .coverage + find . -type d -name __pycache__ -exec rm -rf {} + + find . -type f -name '*.pyc' -delete + +build: clean + python -m build + +publish-test: build + twine upload --repository testpypi dist/* + +publish: build + twine upload dist/* + +all: format lint type-check test diff --git a/README.md b/README.md index 5d4e044..5b08d39 100644 --- a/README.md +++ b/README.md @@ -1,45 +1,294 @@ -# unifi-client +# UniFi Client Python -A Python client library for the UniFi Network Controller API. +A Python client library for the [UniFi Site Manager API](https://developer.ui.com/site-manager-api/gettingstarted). + +## Features + +- ๐Ÿ” Automatic session management with configurable TTL +- ๐Ÿ”„ Automatic retry on authentication failures +- ๐Ÿงต Thread-safe session handling +- โœ… Comprehensive input validation +- ๐Ÿ“ Type hints for better IDE support +- ๐Ÿ Context manager support for proper resource cleanup +- ๐ŸŽฏ Clean, intuitive API ## Installation ```bash -pip install unifi-client +pip install unifi-client-python +``` + +### Development Installation + +```bash +git clone https://github.com/diegofhdz/unifi-client-python.git +cd unifi-client-python +pip install -e ".[dev]" +``` + +## Quick Start + +```python +from unifi_client import UniFiApiClient + +# Initialize the client +client = UniFiApiClient(api_key="your-api-key-here") + +# List all hosts +hosts = client.list_hosts(page_size=20) +print(hosts) + +# Get a specific host +host = client.get_host_by_id("host-id") +print(host) + +# List sites +sites = client.list_sites() +print(sites) + +# Always close the client when done +client.close() +``` + +### Using Context Manager (Recommended) + +```python +from unifi_client import UniFiApiClient + +with UniFiApiClient(api_key="your-api-key-here") as client: + hosts = client.list_hosts() + print(hosts) +# Client is automatically closed +``` + +## API Reference + +### Initialization + +```python +client = UniFiApiClient( + api_key="your-api-key", + api_version="v1", # Optional, default: "v1" + timeout=30, # Optional, default: 30 seconds + session_ttl_minutes=55 # Optional, default: 55 minutes +) +``` + +### Available Methods + +#### Hosts + +```python +# List hosts with pagination +hosts = client.list_hosts(page_size=10, next_token=None) + +# Get host by ID +host = client.get_host_by_id("host-id") +``` + +#### Sites + +```python +# List sites +sites = client.list_sites(page_size=10, next_token=None) +``` + +#### Devices + +```python +# List devices with optional filters +devices = client.list_devices( + time="2024-03-15T14:30:45.123Z", # Optional RFC3339 timestamp + host_ids=["host1", "host2"], # Optional list of host IDs + page_size=10, + next_token=None +) +``` + +#### ISP Metrics + +```python +# Get ISP metrics with duration +metrics = client.get_isp_metrics( + type="5m", # "5m" or "1h" + duration="24h" # "24h", "7d", or "30d" +) + +# Get ISP metrics with timestamp range +metrics = client.get_isp_metrics( + type="1h", + begin_timestamp="2024-03-15T00:00:00.000Z", + end_timestamp="2024-03-15T23:59:59.999Z" +) + +# Query ISP metrics with filters +metrics = client.query_isp_metrics( + type="5m", + site_ids=["site1", "site2"], + host_ids=["host1"], + duration="24h" +) +``` + +#### SD-WAN Configurations + +```python +# List SD-WAN configurations +configs = client.list_sd_wan_configs() + +# Get SD-WAN configuration by ID +config = client.get_sd_wan_config_by_id("config-id") + +# Get SD-WAN configuration status +status = client.get_sd_wan_config_status("config-id") +``` + +## Error Handling + +The client raises `UniFiApiError` for API-related errors: + +```python +from unifi_client import UniFiApiClient, UniFiApiError + +try: + with UniFiApiClient(api_key="your-api-key") as client: + hosts = client.list_hosts() +except UniFiApiError as e: + print(f"API Error: {e}") +except ValueError as e: + print(f"Validation Error: {e}") +``` + +## Advanced Features + +### Session Management + +The client automatically manages sessions with configurable TTL: + +```python +client = UniFiApiClient( + api_key="your-api-key", + session_ttl_minutes=30 # Sessions refresh after 30 minutes +) + +# Manually refresh session if needed +client.refresh_session() +``` + +### Automatic Retry + +The client automatically retries requests once on 401/403 authentication errors after refreshing the session. + +### Thread Safety + +Session access is thread-safe using locks, making it safe to use the same client instance across multiple threads. + +## Testing + +Run tests with pytest: + +```bash +# Run all tests +pytest + +# Run with coverage +pytest --cov=unifi_client --cov-report=html + +# Run specific test file +pytest tests/test_unifi.py + +# Run specific test +pytest tests/test_unifi.py::TestListHosts::test_list_hosts_default_params ``` ## Development -### Setup +### Setup Development Environment ```bash # Clone the repository -git clone https://github.com/diegofhdz/unifi-client-python.git +git clone https://github.com/yourusername/unifi-client-python.git cd unifi-client-python +# Create virtual environment +python -m venv .venv +source .venv/bin/activate # On Windows: .venv\Scripts\activate + # Install in development mode with dev dependencies pip install -e ".[dev]" ``` -### Running Tests +### Code Formatting ```bash -pytest +# Format code with black +black src/ tests/ + +# Check code style with flake8 +flake8 src/ tests/ + +# Type checking with mypy +mypy src/ ``` -### Linting +### Building the Package ```bash -ruff check . -ruff format . +# Install build tools +pip install build twine + +# Build the package +python -m build + +# Check the distribution +twine check dist/* ``` -### Type Checking +### Publishing to PyPI ```bash -mypy src +# Test PyPI (recommended first) +twine upload --repository testpypi dist/* + +# Production PyPI +twine upload dist/* ``` +## Requirements + +- Python >= 3.8 +- requests >= 2.31.0 + ## License -MIT License - see [LICENSE](LICENSE) for details. +MIT License - see LICENSE file for details. + +## Contributing + +Contributions are welcome! Please feel free to submit a Pull Request. + +1. Fork the repository +2. Create your feature branch (`git checkout -b feature/amazing-feature`) +3. Commit your changes (`git commit -m 'Add some amazing feature'`) +4. Push to the branch (`git push origin feature/amazing-feature`) +5. Open a Pull Request + +## Changelog + +### 0.1.0 (Unreleased) + +- Initial release +- Support for Hosts, Sites, Devices, ISP Metrics, and SD-WAN endpoints +- Automatic session management +- Thread-safe implementation +- Comprehensive test coverage + +## Links + +- [UniFi Site Manager API Documentation](https://developer.ui.com/site-manager-api/gettingstarted) +- [GitHub Repository](https://github.com/diegofhdz/unifi-client-python) +- [Issue Tracker](https://github.com/diegofhdz/unifi-client-python/issues) + +## Support + +For bugs, feature requests, or questions, please [open an issue](https://github.com/diegofhdz/unifi-client-python/issues) on GitHub. diff --git a/pyproject.toml b/pyproject.toml index 672b34f..2ba909c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,60 +3,91 @@ requires = ["setuptools>=61.0", "wheel"] build-backend = "setuptools.build_meta" [project] -name = "unifi-client" +name = "unifi-client-python" version = "0.1.0" -description = "A Python client library for the UniFi Network Controller API" +description = "A Python client for the UniFi Site Manager API" readme = "README.md" -license = "MIT" -requires-python = ">=3.9" +requires-python = ">=3.8" +license = {text = "MIT"} authors = [ - {name = "Diego Hernandez"} + {name = "Diego Hernandez", email = "diego.hdz6263@gmail.com"} ] -keywords = ["unifi", "ubiquiti", "network", "api", "client"] classifiers = [ "Development Status :: 3 - Alpha", "Intended Audience :: Developers", - "Operating System :: OS Independent", + "Topic :: Software Development :: Libraries :: Python Modules", + "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", - "Programming Language :: Python :: 3.13", - "Typing :: Typed", ] -dependencies = [] +dependencies = [ + "requests>=2.31.0", +] [project.optional-dependencies] dev = [ - "pytest>=7.0.0", - "pytest-cov>=4.0.0", - "ruff>=0.1.0", - "mypy>=1.0.0", + "pytest>=7.4.0", + "pytest-cov>=4.1.0", + "pytest-mock>=3.11.1", + "black>=23.7.0", + "flake8>=6.1.0", + "mypy>=1.5.0", + "types-requests>=2.31.0", ] [project.urls] Homepage = "https://github.com/diegofhdz/unifi-client-python" +Documentation = "https://github.com/diegofhdz/unifi-client-python#readme" Repository = "https://github.com/diegofhdz/unifi-client-python" -Issues = "https://github.com/diegofhdz/unifi-client-python/issues" +"Bug Tracker" = "https://github.com/diegofhdz/unifi-client-python/issues" -[tool.setuptools.packages.find] -where = ["src"] +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +addopts = [ + "--verbose", + "--cov=unifi_client", + "--cov-report=term-missing", + "--cov-report=html", +] -[tool.ruff] +[tool.black] line-length = 88 -target-version = "py39" - -[tool.ruff.lint] -select = ["E", "F", "I", "N", "W", "UP"] +target-version = ['py38', 'py39', 'py310', 'py311'] +include = '\.pyi?$' +extend-exclude = ''' +/( + # directories + \.eggs + | \.git + | \.hg + | \.mypy_cache + | \.tox + | \.venv + | build + | dist +)/ +''' [tool.mypy] -python_version = "3.9" +python_version = "3.8" warn_return_any = true warn_unused_configs = true -strict = true +disallow_untyped_defs = true +disallow_incomplete_defs = true +check_untyped_defs = true +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = true +warn_no_return = true +strict_equality = true -[tool.pytest.ini_options] -testpaths = ["tests"] -python_files = ["test_*.py"] -python_functions = ["test_*"] +[[tool.mypy.overrides]] +module = "tests.*" +disallow_untyped_defs = false \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..fb79401 --- /dev/null +++ b/setup.py @@ -0,0 +1,44 @@ +from setuptools import setup, find_packages + +with open("README.md", "r", encoding="utf-8") as fh: + long_description = fh.read() + +setup( + name="unifi-client-python", + version="0.1.0", + author="Diego Hernandez", + author_email="diego.hdz6263@gmail.com", + description="A Python client for the UniFi Site Manager API", + long_description=long_description, + long_description_content_type="text/markdown", + url="https://github.com/diegofhdz/unifi-client-python", + packages=find_packages(where="src"), + package_dir={"": "src"}, + classifiers=[ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Topic :: Software Development :: Libraries :: Python Modules", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + ], + python_requires=">=3.8", + install_requires=[ + "requests>=2.31.0", + ], + extras_require={ + "dev": [ + "pytest>=7.4.0", + "pytest-cov>=4.1.0", + "pytest-mock>=3.11.1", + "black>=23.7.0", + "flake8>=6.1.0", + "mypy>=1.5.0", + "types-requests>=2.31.0", + ], + }, +) \ No newline at end of file diff --git a/src/unifi_client/__init__.py b/src/unifi_client/__init__.py index 8875d34..17ba5c2 100644 --- a/src/unifi_client/__init__.py +++ b/src/unifi_client/__init__.py @@ -1,7 +1,8 @@ -"""UniFi Python Client Library. - -A Python client library for interacting with the UniFi Network Controller API. """ +UniFi Client Python - A Python client for the UniFi Site Manager API +""" + +from unifi_client.unifi import UniFiApiClient, UniFiApiError __version__ = "0.1.0" -__all__ = ["__version__"] +__all__ = ["UniFiApiClient", "UniFiApiError"] \ No newline at end of file diff --git a/src/unifi_client/py.typed b/src/unifi_client/py.typed deleted file mode 100644 index e69de29..0000000 diff --git a/src/unifi_client/unifi.py b/src/unifi_client/unifi.py index 762c7eb..244d853 100644 --- a/src/unifi_client/unifi.py +++ b/src/unifi_client/unifi.py @@ -1,15 +1,17 @@ import requests -from typing import Optional, Any import logging from datetime import datetime, timedelta from threading import Lock +from requests.adapters import HTTPAdapter +from typing import Optional, Any, Dict, List logger = logging.getLogger(__name__) class UniFiApiError(Exception): - """Custom exception for UniFi API errors""" - + """ + Custom exception for UniFi API errors + """ pass @@ -67,7 +69,7 @@ def _create_session(self) -> requests.Session: } ) - adapter = requests.adapters.HTTPAdapter( + adapter = HTTPAdapter( pool_connections=10, pool_maxsize=20, max_retries=3, pool_block=False ) session.mount("https://", adapter) @@ -84,60 +86,394 @@ def refresh_session(self) -> None: self._session_created_at = None logger.info("Session manually refreshed") - def list_hosts( - self, page_size: int = 10, next_token: Optional[str] = None - ) -> dict[str, str]: + def _make_request( + self, + method: str, + endpoint: str, + params: Optional[dict] = None, + **kwargs + ) -> Dict[str, Any]: """ - List UniFi hosts with pagination support. + Centralized request handler with automatic retry on auth failures. Args: - page_size: Number of results per page (1-100) - next_token: Token for pagination + method: HTTP method (GET, POST, etc.) + endpoint: API endpoint path (without base URL) + params: Query parameters + **kwargs: Additional arguments to pass to requests Returns: - Parsed JSON response as dictionary + Parsed JSON response Raises: UniFiApiError: If the API request fails - ValueError: If page_size is invalid """ - if not 1 <= page_size <= 100: - raise ValueError("page_size must be between 1 and 100") - - url = f"{self.base_url}/hosts" - params = {"pageSize": str(page_size)} + url = f"{self.base_url}/{endpoint}" - if next_token: - params["nextToken"] = next_token - - try: - response = self.session.get() - url=url, params=params, timeout=self.timeout + def _attempt_request(): + response = self.session.request( + method=method, + url=url, + params=params, + timeout=self.timeout, + **kwargs ) response.raise_for_status() return response.json() + try: + return _attempt_request() + except requests.HTTPError as e: - # If 401/403, might be auth issue - try refresh once + # Retry once on auth errors if e.response.status_code in (401, 403): - logger.warning("Authentication error, attempting session refresh") + logger.warning(f"Authentication error (HTTP {e.response.status_code}), refreshing session") self.refresh_session() - response = self.session.get(url, params=params, timeout=self.timeout) - response.raise_for_status() - return response.json() + try: + return _attempt_request() + except requests.HTTPError as retry_error: + logger.error(f"Retry failed: {retry_error.response.status_code} - {retry_error.response.text}") + raise UniFiApiError(f"API request failed after retry: {retry_error.response.status_code}") from retry_error logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}") - raise UniFiApiError(f"API request failed: {e.response.status_code}") + raise UniFiApiError(f"API request failed: {e.response.status_code}") from e except requests.Timeout: logger.error(f"Request timed out after {self.timeout} seconds") - raise UniFiApiError("Request timed out") + raise UniFiApiError(f"Request timed out after {self.timeout} seconds") + except requests.RequestException as e: logger.error(f"Request failed: {str(e)}") - raise UniFiApiError(f"Request failed: {str(e)}") + raise UniFiApiError(f"Request failed: {str(e)}") from e + except ValueError as e: logger.error(f"Invalid JSON response: {str(e)}") - raise UniFiApiError("Invalid JSON response from API") + raise UniFiApiError("Invalid JSON response from API") from e + + def _validate_page_size(self, page_size: int) -> None: + """Validate page_size parameter""" + if not 1 <= page_size <= 100: + raise ValueError("page_size must be between 1 and 100") + + def _validate_rfc3339(self, timestamp: str) -> datetime: + """ + Validate and parse RFC3339 timestamp. + + Args: + timestamp: RFC3339 formatted timestamp string + + Returns: + Parsed datetime object + + Raises: + ValueError: If timestamp format is invalid + """ + try: + # Handle both Z and timezone offset formats + if timestamp.endswith('Z'): + return datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%fZ") + else: + # Remove colon from timezone for strptime + timestamp_clean = timestamp[:-3] + timestamp[-2:] + return datetime.strptime(timestamp_clean, "%Y-%m-%dT%H:%M:%S.%f%z") + except ValueError as e: + raise ValueError(f"Invalid RFC3339 timestamp format: {timestamp}. Expected format: YYYY-MM-DDTHH:MM:SS.sssZ or YYYY-MM-DDTHH:MM:SS.sssยฑHH:MM") from e + + def _validate_timestamp_range( + self, + begin_timestamp: Optional[str], + end_timestamp: Optional[str] + ) -> None: + """ + Validate that end_timestamp > begin_timestamp. + + Args: + begin_timestamp: Start timestamp in RFC3339 format + end_timestamp: End timestamp in RFC3339 format + + Raises: + ValueError: If end_timestamp is not greater than begin_timestamp + """ + if not (begin_timestamp and end_timestamp): + return + + begin_dt = self._validate_rfc3339(begin_timestamp) + end_dt = self._validate_rfc3339(end_timestamp) + + if end_dt <= begin_dt: + raise ValueError("'end_timestamp' must be strictly greater than 'begin_timestamp'") + + def list_hosts( + self, + page_size: int = 10, + next_token: Optional[str] = None + ) -> Dict[str, Any]: + """ + List UniFi hosts with pagination support. + + Args: + page_size: Number of results per page (1-100) + next_token: Token for pagination + + Returns: + Parsed JSON response as dictionary + + Raises: + UniFiApiError: If the API request fails + ValueError: If page_size is invalid + """ + self._validate_page_size(page_size) + params = {"pageSize": str(page_size)} + if next_token: + params["nextToken"] = next_token + + return self._make_request("GET", "hosts", params=params) + + def get_host_by_id(self, host_id: str) -> Dict[str, Any]: + """ + Retrieves detailed information about a specific host by ID. + + Args: + host_id: Unique identifier of the host + + Returns: + Parsed JSON response as dictionary + + Raises: + UniFiApiError: If the API request fails + ValueError: If host_id is empty + """ + if not host_id: + raise ValueError("host_id cannot be empty") + + return self._make_request("GET", f"hosts/{host_id}") + + def list_sites( + self, + page_size: int = 10, + next_token: Optional[str] = None + ) -> Dict[str, Any]: + """ + Retrieves a list of all sites (from hosts running the UniFi Network application) + associated with the UI account making the API call. + + Args: + page_size: Number of results per page (1-100) + next_token: Token for pagination + + Returns: + Parsed JSON response as dictionary + + Raises: + UniFiApiError: If the API request fails + ValueError: If page_size is invalid + """ + self._validate_page_size(page_size) + params = {"pageSize": str(page_size)} + if next_token: + params["nextToken"] = next_token + + return self._make_request("GET", "sites", params=params) + + def list_devices( + self, + time: Optional[str] = None, + host_ids: Optional[List[str]] = None, + page_size: int = 10, + next_token: Optional[str] = None + ) -> Dict[str, Any]: + """ + Retrieves a list of UniFi devices managed by hosts where the UI account + making the API call is the owner or a super admin. + + Args: + time: Last processed timestamp of devices in RFC3339 format. Example: 2025-06-17T02:45:58Z + host_ids: List of host IDs to filter the results + page_size: Number of results per page (1-100) + next_token: Token for pagination + + Returns: + Parsed JSON response as dictionary + + Raises: + UniFiApiError: If the API request fails + ValueError: If page_size is invalid or time format is invalid + """ + self._validate_page_size(page_size) + if time: + self._validate_rfc3339(time) + + params = {"pageSize": str(page_size)} + if next_token: + params["nextToken"] = next_token + if time: + params["time"] = time + if host_ids: + # Adjust format based on actual API specification + params["hostIds"] = ",".join(host_ids) + + return self._make_request("GET", "devices", params=params) + + def get_isp_metrics( + self, + type: str = "5m", + begin_timestamp: Optional[str] = None, + end_timestamp: Optional[str] = None, + duration: Optional[str] = None + ) -> Dict[str, Any]: + """ + Retrieves ISP metrics data for all sites linked to the UI account's API key. + 5-minute interval metrics are available for at least 24 hours, and 1-hour + interval metrics are available for at least 30 days. + + Args: + type: Specifies whether metrics are returned using 5m or 1h intervals + begin_timestamp: The earliest timestamp to retrieve data from (RFC3339 format) + end_timestamp: The latest timestamp to retrieve data up to (RFC3339 format) + duration: Specifies the time range of metrics to retrieve, starting from when + the request is made. Supports 24h for 5-minute metrics, and 7d or 30d + for 1-hour metrics. Cannot be used with beginTimestamp or endTimestamp. + + Returns: + Parsed JSON response as dictionary + + Raises: + UniFiApiError: If the API request fails + ValueError: If parameters are invalid + """ + # Validation + if type not in ("5m", "1h"): + raise ValueError("'type' parameter must be either '5m' or '1h'") + + if duration and (begin_timestamp or end_timestamp): + raise ValueError("'duration' cannot be used with begin_timestamp or end_timestamp") + + # Validate and compare timestamps + if begin_timestamp or end_timestamp: + self._validate_timestamp_range(begin_timestamp, end_timestamp) + + # Build params + params = {} + if duration: + params["duration"] = duration + if begin_timestamp: + params["beginTimestamp"] = begin_timestamp + if end_timestamp: + params["endTimestamp"] = end_timestamp + + return self._make_request("GET", f"ea/isp-metrics/{type}", params=params) + + def query_isp_metrics( + self, + type: str = "5m", + begin_timestamp: Optional[str] = None, + end_timestamp: Optional[str] = None, + duration: Optional[str] = None, + site_ids: Optional[List[str]] = None, + host_ids: Optional[List[str]] = None + ) -> Dict[str, Any]: + """ + Retrieves ISP metrics data based on specific query parameters. + 5-minute interval metrics are available for at least 24 hours, and 1-hour + interval metrics are available for at least 30 days. + + Note: If the UI account lacks access to all requested sites, a 502 error is returned. + If partial access is granted, the response will include status: partialSuccess. + + Args: + type: Specifies whether metrics are returned using 5m or 1h intervals + begin_timestamp: The earliest timestamp to retrieve data from (RFC3339 format) + end_timestamp: The latest timestamp to retrieve data up to (RFC3339 format) + duration: Specifies the time range of metrics to retrieve, starting from when + the request is made. Supports 24h for 5-minute metrics, and 7d or 30d + for 1-hour metrics. Cannot be used with beginTimestamp or endTimestamp. + site_ids: List of site IDs to filter the results + host_ids: List of host IDs to filter the results + + Returns: + Parsed JSON response as dictionary + + Raises: + UniFiApiError: If the API request fails + ValueError: If parameters are invalid + """ + # Validation + if type not in ("5m", "1h"): + raise ValueError("'type' parameter must be either '5m' or '1h'") + + if duration and (begin_timestamp or end_timestamp): + raise ValueError("'duration' cannot be used with begin_timestamp or end_timestamp") + + # Validate and compare timestamps + if begin_timestamp or end_timestamp: + self._validate_timestamp_range(begin_timestamp, end_timestamp) + + # Build request body + body = {} + if duration: + body["duration"] = duration + if begin_timestamp: + body["beginTimestamp"] = begin_timestamp + if end_timestamp: + body["endTimestamp"] = end_timestamp + if site_ids: + body["siteIds"] = site_ids + if host_ids: + body["hostIds"] = host_ids + + return self._make_request("POST", f"ea/isp-metrics/{type}/query", json=body) + + def list_sd_wan_configs(self) -> Dict[str, Any]: + """ + Retrieves a list of all SD-WAN configurations associated with the UI account + making the API call. + + Returns: + Parsed JSON response as dictionary + + Raises: + UniFiApiError: If the API request fails + """ + return self._make_request("GET", "ea/sd-wan-configs") + + def get_sd_wan_config_by_id(self, config_id: str) -> Dict[str, Any]: + """ + Retrieves detailed information about a specific SD-WAN configuration by ID. + + Args: + config_id: Unique identifier of the SD-WAN configuration + + Returns: + Parsed JSON response as dictionary + + Raises: + UniFiApiError: If the API request fails + ValueError: If config_id is empty + """ + if not config_id: + raise ValueError("config_id cannot be empty") + + return self._make_request("GET", f"ea/sd-wan-configs/{config_id}") + + def get_sd_wan_config_status(self, config_id: str) -> Dict[str, Any]: + """ + Retrieves the status of a specific SD-WAN configuration, including deployment + progress, errors, and associated hubs. + + Args: + config_id: Unique identifier of the SD-WAN configuration + + Returns: + Parsed JSON response as dictionary + + Raises: + UniFiApiError: If the API request fails + ValueError: If config_id is empty + """ + if not config_id: + raise ValueError("config_id cannot be empty") + + return self._make_request("GET", f"ea/sd-wan-configs/{config_id}/status") def close(self) -> None: """Close the session""" @@ -154,5 +490,4 @@ def __exit__(self, exc_type, exc_val, exc_tb): self.close() def __del__(self): - self.close() - + self.close() \ No newline at end of file diff --git a/tests/test_init.py b/tests/test_init.py index 6f002b4..b9f883f 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -6,8 +6,3 @@ def test_version() -> None: """Test that the package version is defined.""" assert unifi_client.__version__ == "0.1.0" - - -def test_version_in_all() -> None: - """Test that __version__ is exported in __all__.""" - assert "__version__" in unifi_client.__all__ diff --git a/tests/test_unifi.py b/tests/test_unifi.py new file mode 100644 index 0000000..8d6cadd --- /dev/null +++ b/tests/test_unifi.py @@ -0,0 +1,434 @@ +import pytest +from unittest.mock import Mock, patch +from datetime import datetime, timedelta +import requests +from unifi_client.unifi import UniFiApiClient, UniFiApiError + + +class TestUniFiApiClientInit: + """Test client initialization""" + + def test_init_with_valid_api_key(self): + client = UniFiApiClient(api_key="test-api-key") + assert client.api_key == "test-api-key" + assert client.api_version == "v1" + assert client.base_url == "https://api.ui.com/v1" + assert client.timeout == 30 + + def test_init_with_custom_params(self): + client = UniFiApiClient( + api_key="test-key", + api_version="v2", + timeout=60, + session_ttl_minutes=30 + ) + assert client.api_version == "v2" + assert client.timeout == 60 + assert client.session_ttl == timedelta(minutes=30) + + def test_init_with_empty_api_key_raises_error(self): + with pytest.raises(ValueError, match="API key cannot be empty"): + UniFiApiClient(api_key="") + + def test_init_with_none_api_key_raises_error(self): + with pytest.raises(ValueError, match="API key cannot be empty"): + UniFiApiClient(api_key=None) + + +class TestSessionManagement: + """Test session creation and management""" + + def test_session_creates_new_session(self): + client = UniFiApiClient(api_key="test-key") + session = client.session + assert session is not None + assert isinstance(session, requests.Session) + assert session.headers["X-API-Key"] == "test-key" + assert session.headers["Accept"] == "application/json" + + def test_session_reuses_existing_session(self): + client = UniFiApiClient(api_key="test-key") + session1 = client.session + session2 = client.session + assert session1 is session2 + + def test_session_refreshes_after_ttl(self): + client = UniFiApiClient(api_key="test-key", session_ttl_minutes=0) + session1 = client.session + # Force time to pass + client._session_created_at = datetime.now() - timedelta(minutes=1) + session2 = client.session + assert session1 is not session2 + + def test_refresh_session_closes_old_session(self): + client = UniFiApiClient(api_key="test-key") + old_session = client.session + old_session.close = Mock() + + client.refresh_session() + + old_session.close.assert_called_once() + assert client._session is None + + def test_context_manager_closes_session(self): + with UniFiApiClient(api_key="test-key") as client: + session = client.session + session.close = Mock() + + session.close.assert_called() + + +class TestValidation: + """Test validation helper methods""" + + def test_validate_rfc3339_with_z_suffix(self): + client = UniFiApiClient(api_key="test-key") + timestamp = "2024-03-15T14:30:45.123Z" + result = client._validate_rfc3339(timestamp) + assert isinstance(result, datetime) + assert result.year == 2024 + assert result.month == 3 + assert result.day == 15 + + def test_validate_rfc3339_with_timezone_offset(self): + client = UniFiApiClient(api_key="test-key") + timestamp = "2024-03-15T14:30:45.123+05:30" + result = client._validate_rfc3339(timestamp) + assert isinstance(result, datetime) + + def test_validate_rfc3339_with_invalid_format_raises_error(self): + client = UniFiApiClient(api_key="test-key") + with pytest.raises(ValueError, match="Invalid RFC3339 timestamp format"): + client._validate_rfc3339("2024-03-15") + + def test_validate_timestamp_range_valid(self): + client = UniFiApiClient(api_key="test-key") + begin = "2024-03-15T10:00:00.000Z" + end = "2024-03-15T14:00:00.000Z" + # Should not raise + client._validate_timestamp_range(begin, end) + + def test_validate_timestamp_range_invalid_raises_error(self): + client = UniFiApiClient(api_key="test-key") + begin = "2024-03-15T14:00:00.000Z" + end = "2024-03-15T10:00:00.000Z" + with pytest.raises(ValueError, match="must be strictly greater"): + client._validate_timestamp_range(begin, end) + + def test_validate_timestamp_range_with_none_values(self): + client = UniFiApiClient(api_key="test-key") + # Should not raise + client._validate_timestamp_range(None, None) + client._validate_timestamp_range("2024-03-15T10:00:00.000Z", None) + client._validate_timestamp_range(None, "2024-03-15T14:00:00.000Z") + + +class TestMakeRequest: + """Test centralized request handling""" + + @patch('unifi_client.unifi.requests.Session.request') + def test_make_request_success(self, mock_request): + mock_response = Mock() + mock_response.json.return_value = {"data": "test"} + mock_response.raise_for_status = Mock() + mock_request.return_value = mock_response + + client = UniFiApiClient(api_key="test-key") + result = client._make_request("GET", "hosts") + + assert result == {"data": "test"} + mock_request.assert_called_once() + + @patch('unifi_client.unifi.requests.Session.request') + def test_make_request_with_params(self, mock_request): + mock_response = Mock() + mock_response.json.return_value = {"data": "test"} + mock_response.raise_for_status = Mock() + mock_request.return_value = mock_response + + client = UniFiApiClient(api_key="test-key") + params = {"pageSize": "10"} + client._make_request("GET", "hosts", params=params) + + call_args = mock_request.call_args + assert call_args.kwargs["params"] == params + + @patch('unifi_client.unifi.requests.Session.request') + def test_make_request_retries_on_401(self, mock_request): + # First call returns 401, second call succeeds + error_response = Mock() + error_response.status_code = 401 + error_response.text = "Unauthorized" + + success_response = Mock() + success_response.json.return_value = {"data": "test"} + success_response.raise_for_status = Mock() + + mock_request.side_effect = [ + requests.HTTPError(response=error_response), + success_response + ] + + client = UniFiApiClient(api_key="test-key") + with patch.object(client, 'refresh_session'): + result = client._make_request("GET", "hosts") + + assert result == {"data": "test"} + assert mock_request.call_count == 2 + + @patch('unifi_client.unifi.requests.Session.request') + def test_make_request_timeout_raises_error(self, mock_request): + mock_request.side_effect = requests.Timeout() + + client = UniFiApiClient(api_key="test-key") + with pytest.raises(UniFiApiError, match="timed out"): + client._make_request("GET", "hosts") + + @patch('unifi_client.unifi.requests.Session.request') + def test_make_request_http_error_raises_unified_error(self, mock_request): + error_response = Mock() + error_response.status_code = 500 + error_response.text = "Internal Server Error" + mock_request.side_effect = requests.HTTPError(response=error_response) + + client = UniFiApiClient(api_key="test-key") + with pytest.raises(UniFiApiError, match="API request failed: 500"): + client._make_request("GET", "hosts") + + @patch('unifi_client.unifi.requests.Session.request') + def test_make_request_invalid_json_raises_error(self, mock_request): + mock_response = Mock() + mock_response.raise_for_status = Mock() + mock_response.json.side_effect = ValueError("Invalid JSON") + mock_request.return_value = mock_response + + client = UniFiApiClient(api_key="test-key") + with pytest.raises(UniFiApiError, match="Invalid JSON response"): + client._make_request("GET", "hosts") + + +class TestListHosts: + """Test list_hosts endpoint""" + + @patch.object(UniFiApiClient, '_make_request') + def test_list_hosts_default_params(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + result = client.list_hosts() + + mock_request.assert_called_once_with( + "GET", "hosts", params={"pageSize": "10"} + ) + assert result == {"data": []} + + @patch.object(UniFiApiClient, '_make_request') + def test_list_hosts_with_custom_page_size(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + client.list_hosts(page_size=50) + + call_args = mock_request.call_args + assert call_args[1]["params"]["pageSize"] == "50" + + @patch.object(UniFiApiClient, '_make_request') + def test_list_hosts_with_next_token(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + client.list_hosts(next_token="token123") + + call_args = mock_request.call_args + assert call_args[1]["params"]["nextToken"] == "token123" + + def test_list_hosts_invalid_page_size_raises_error(self): + client = UniFiApiClient(api_key="test-key") + with pytest.raises(ValueError, match="page_size must be between 1 and 100"): + client.list_hosts(page_size=101) + + with pytest.raises(ValueError, match="page_size must be between 1 and 100"): + client.list_hosts(page_size=0) + + +class TestGetHostById: + """Test get_host_by_id endpoint""" + + @patch.object(UniFiApiClient, '_make_request') + def test_get_host_by_id_success(self, mock_request): + mock_request.return_value = {"id": "host123", "name": "Test Host"} + + client = UniFiApiClient(api_key="test-key") + result = client.get_host_by_id("host123") + + mock_request.assert_called_once_with("GET", "hosts/host123") + assert result["id"] == "host123" + + def test_get_host_by_id_empty_id_raises_error(self): + client = UniFiApiClient(api_key="test-key") + with pytest.raises(ValueError, match="host_id cannot be empty"): + client.get_host_by_id("") + + +class TestListSites: + """Test list_sites endpoint""" + + @patch.object(UniFiApiClient, '_make_request') + def test_list_sites_success(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + result = client.list_sites() + + mock_request.assert_called_once_with( + "GET", "sites", params={"pageSize": "10"} + ) + + +class TestListDevices: + """Test list_devices endpoint""" + + @patch.object(UniFiApiClient, '_make_request') + def test_list_devices_with_time_filter(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + timestamp = "2024-03-15T14:30:45.123Z" + client.list_devices(time=timestamp) + + call_args = mock_request.call_args + assert call_args[1]["params"]["time"] == timestamp + + @patch.object(UniFiApiClient, '_make_request') + def test_list_devices_with_host_ids(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + client.list_devices(host_ids=["host1", "host2"]) + + call_args = mock_request.call_args + assert call_args[1]["params"]["hostIds"] == "host1,host2" + + def test_list_devices_with_invalid_time_raises_error(self): + client = UniFiApiClient(api_key="test-key") + with pytest.raises(ValueError, match="Invalid RFC3339 timestamp"): + client.list_devices(time="invalid-timestamp") + + +class TestGetIspMetrics: + """Test get_isp_metrics endpoint""" + + @patch.object(UniFiApiClient, '_make_request') + def test_get_isp_metrics_with_duration(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + client.get_isp_metrics(type="5m", duration="24h") + + call_args = mock_request.call_args + assert call_args[1]["params"]["duration"] == "24h" + + @patch.object(UniFiApiClient, '_make_request') + def test_get_isp_metrics_with_timestamps(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + begin = "2024-03-15T10:00:00.000Z" + end = "2024-03-15T14:00:00.000Z" + client.get_isp_metrics(begin_timestamp=begin, end_timestamp=end) + + call_args = mock_request.call_args + assert call_args[1]["params"]["beginTimestamp"] == begin + assert call_args[1]["params"]["endTimestamp"] == end + + def test_get_isp_metrics_invalid_type_raises_error(self): + client = UniFiApiClient(api_key="test-key") + with pytest.raises(ValueError, match="must be either '5m' or '1h'"): + client.get_isp_metrics(type="10m") + + def test_get_isp_metrics_duration_with_timestamps_raises_error(self): + client = UniFiApiClient(api_key="test-key") + with pytest.raises(ValueError, match="cannot be used with"): + client.get_isp_metrics( + duration="24h", + begin_timestamp="2024-03-15T10:00:00.000Z" + ) + + +class TestQueryIspMetrics: + """Test query_isp_metrics endpoint""" + + @patch.object(UniFiApiClient, '_make_request') + def test_query_isp_metrics_with_filters(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + client.query_isp_metrics( + type="5m", + site_ids=["site1", "site2"], + host_ids=["host1"] + ) + + call_args = mock_request.call_args + assert call_args[0] == ("POST", "ea/isp-metrics/5m/query") + assert "siteIds" in call_args[1]["json"] + assert "hostIds" in call_args[1]["json"] + + +class TestSdWanMethods: + """Test SD-WAN related endpoints""" + + @patch.object(UniFiApiClient, '_make_request') + def test_list_sd_wan_configs(self, mock_request): + mock_request.return_value = {"data": []} + + client = UniFiApiClient(api_key="test-key") + result = client.list_sd_wan_configs() + + mock_request.assert_called_once_with("GET", "ea/sd-wan-configs") + + @patch.object(UniFiApiClient, '_make_request') + def test_get_sd_wan_config_by_id(self, mock_request): + mock_request.return_value = {"id": "config123"} + + client = UniFiApiClient(api_key="test-key") + result = client.get_sd_wan_config_by_id("config123") + + mock_request.assert_called_once_with("GET", "ea/sd-wan-configs/config123") + + def test_get_sd_wan_config_by_id_empty_raises_error(self): + client = UniFiApiClient(api_key="test-key") + with pytest.raises(ValueError, match="config_id cannot be empty"): + client.get_sd_wan_config_by_id("") + + @patch.object(UniFiApiClient, '_make_request') + def test_get_sd_wan_config_status(self, mock_request): + mock_request.return_value = {"status": "active"} + + client = UniFiApiClient(api_key="test-key") + result = client.get_sd_wan_config_status("config123") + + mock_request.assert_called_once_with("GET", "ea/sd-wan-configs/config123/status") + + +class TestCleanup: + """Test resource cleanup""" + + def test_close_closes_session(self): + client = UniFiApiClient(api_key="test-key") + session = client.session + session.close = Mock() + + client.close() + + session.close.assert_called_once() + assert client._session is None + + def test_del_closes_session(self): + client = UniFiApiClient(api_key="test-key") + session = client.session + session.close = Mock() + + del client + + session.close.assert_called()