diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9f94e5d --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.vscode +__pycache__ diff --git a/API.md b/API.md new file mode 100644 index 0000000..3d685f2 --- /dev/null +++ b/API.md @@ -0,0 +1,189 @@ +# Server Inventory Management API + +A Flask-based REST API for managing server inventory with PostgreSQL backend. + +## Quick Start + +### Prerequisites +- Docker and Docker Compose +- Python 3.9+ (for CLI and tests) + +### Start the Service + +```bash +docker compose up --build -d +``` + +The API will be available at `http://localhost:5001` + +The database is automatically initialized with the schema from `init_db.sql`. + +### Stop the Service + +```bash +docker compose down +``` + +## Running Tests + +### All Tests +```bash +make test # From the root directory +``` + +### Unit Tests +```bash +make unit-test # From the root directory +``` + +![alt text](assets/image.png) + +### Integration Tests +```bash +make integration-test # From the root directory +``` + +![alt text](assets/image-1.png) + +### CLI Tests +```bash +make cli-test # From the root directory +``` + +![alt text](assets/image-2.png) + +## API Endpoints + +Base URL: `http://localhost:5001` + +### List Servers +```bash +GET /servers + +# Example +curl http://localhost:5001/servers + +# Response +[ + { + "id": 1, + "hostname": "web-server-01", + "ip_address": "192.168.1.10", + "server_state": "active" + }, + { + "id": 2, + "hostname": "web-server-02", + "ip_address": "192.168.1.10", + "server_state": "active" + } +] +``` + +### Get Server +```bash +GET /servers/{id} + +# Example +curl http://localhost:5001/servers/1 + +# Response +{ + "id": 1, + "hostname": "web-server-01", + "ip_address": "192.168.1.10", + "server_state": "active" +} +``` + +### Create Server +```bash +POST /servers +Content-Type: application/json + +# Example +curl -X POST http://localhost:5001/servers \ + -H "Content-Type: application/json" \ + -d '{ + "hostname": "web-server-01", + "ip_address": "192.168.1.10", + "server_state": "active" + }' + +# Response (201 Created) +{ + "server_id": 1 +} +``` + +**Validation:** +- `hostname`: Required, must be unique +- `ip_address`: Required, valid IPv4 format +- `server_state`: Required, one of: `active`, `offline`, `retired` + +### Update Server +```bash +PUT /servers/{id} +Content-Type: application/json + +# Example +curl -X PUT http://localhost:5001/servers/1 \ + -H "Content-Type: application/json" \ + -d '{ + "hostname": "updated-hostname", + "ip_address": "10.0.0.1", + "server_state": "offline" + }' + +# Response (200 OK) +{ + "id": 1, + "hostname": "updated-hostname", + "ip_address": "10.0.0.1", + "server_state": "offline" +} +``` + +**Note:** At least one field must be provided. Omitted fields remain unchanged. The updated hostname must also still be unique. + +### Delete Server +```bash +DELETE /servers/{id} + +# Example +curl -X DELETE http://localhost:5001/servers/1 + +# Response (200 OK) +{ + "message": "Server deleted" +} +``` + +## Error Responses + +- `400 Bad Request`: Validation error or invalid input +- `404 Not Found`: Server ID not found +- `500 Internal Server Error`: Server error + +## Development + +### Install Dependencies +```bash +pip install -r requirements.txt +``` + +### Database Schema +The database schema is defined in `init_db.sql` and is automatically applied when the PostgreSQL container starts. + +### Project Structure +``` +. +├── flask_api/ +│ ├── app.py # Flask API application +│ ├── cli.py # CLI tool +| |__ Dockerfile # Docker image file. +│ ├── requirements.txt # Python dependencies +│ └── tests/ # Test suite +├── init_db.sql # Database schema +├── docker-compose.yml # Docker Compose configuration +└── scripts/ # Test scripts diff --git a/CLI.md b/CLI.md new file mode 100644 index 0000000..f1f395c --- /dev/null +++ b/CLI.md @@ -0,0 +1,121 @@ +# Server Inventory Management CLI + +A command-line interface for managing servers in the inventory system via the REST API. + +## Prerequisites + +- The Flask API and Postgres database must be running (see API.md for setup instructions) +- Docker and Docker Compose +- Python 3.9+ (for CLI and tests) + +## Use the CLI + +The CLI connects to the REST API to perform all operations. + +### Basic Usage + +- Launch the API and database +` docker compose up --build -d ` + +- Create a virtual environment and install dependencies + +```bash +python3 -m venv challenge-env +source challenge-env/bin/activate +pip install -r requirements.txt +``` + + +```bash +python flask_api/cli.py list +python flask_api/cli.py get +python flask_api/cli.py create --hostname --ip-address --state +python flask_api/cli.py update [options] +python flask_api/cli.py delete +``` + +### Commands + +#### List Servers +List all servers in the inventory. + +```bash +python flask_api/cli.py list +``` + +#### Get Server +Get details of a specific server by ID. + +```bash +python flask_api/cli.py get 1 +``` + +#### Create Server +Create a new server in the inventory. + +```bash +python flask_api/cli.py create --hostname "web-server-01" --ip-address "192.168.1.10" --state active +``` + +Valid states: `active`, `offline`, `retired` + +#### Update Server +Update an existing server. You can update one or more fields. + +```bash +# Update hostname only +python flask_api/cli.py update 1 --hostname "new-hostname" + +# Update multiple fields +python flask_api/cli.py update 1 --hostname "new-hostname" --ip-address "10.0.0.1" --state offline +``` + +#### Delete Server +Delete a server from the inventory (requires confirmation). + +```bash +python flask_api/cli.py delete 1 +``` + +### Options + +#### Global Options + +- `--api-url `: Specify API base URL (default: http://localhost:5001) + +### Examples + +```bash +# List all servers +python flask_api/cli.py list + +# Create a server +python flask_api/cli.py create --hostname "db-server-01" --ip-address "10.0.0.5" --state active + +# Update server state only +python flask_api/cli.py update 2 --state offline + +# Get server details +python flask_api/cli.py get 1 + +# Delete a server (will prompt for confirmation) +python flask_api/cli.py delete 3 + +# Use custom API URL +python flask_api/cli.py --api-url http://api.example.com:8080 list +``` + +### Error Handling + +The CLI provides clear error messages for: +- Invalid IP address formats +- Invalid server states +- Missing required fields +- Server not found (404) +- Duplicate hostnames +- API connection errors + +### Exit Codes + +- `0`: Success +- `1`: Error (validation error, not found, connection error, etc.) diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..7b3f1a3 --- /dev/null +++ b/Makefile @@ -0,0 +1,11 @@ +unit-test: + ./scripts/unit-test.sh + +integration-test: + ./scripts/integration-test.sh + +cli-test: + ./scripts/cli-test.sh + +test: unit-test integration-test cli-test + diff --git a/README.md b/README.md index 3145d38..673364a 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,5 @@ Requirements: Validate that: - hostname is unique - IP address looks like an IP - -State is one of: active, offline, retired +- State is one of: active, offline, retired diff --git a/assets/image-1.png b/assets/image-1.png new file mode 100644 index 0000000..5ffc0cd Binary files /dev/null and b/assets/image-1.png differ diff --git a/assets/image-2.png b/assets/image-2.png new file mode 100644 index 0000000..98a4bf3 Binary files /dev/null and b/assets/image-2.png differ diff --git a/assets/image.png b/assets/image.png new file mode 100644 index 0000000..23a8fc0 Binary files /dev/null and b/assets/image.png differ diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..786798d --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,24 @@ +services: + postgres: + image: postgres:latest + container_name: postgres_container + environment: + POSTGRES_USER: your_username + POSTGRES_PASSWORD: your_password + POSTGRES_DB: your_database + ports: + - "5432:5432" + volumes: + - ./init_db.sql:/docker-entrypoint-initdb.d/init_db.sql + + flask_api: + build: + context: ./flask_api + container_name: flask_api_container + ports: + - "5001:5000" + depends_on: + - postgres + + + diff --git a/flask_api/Dockerfile b/flask_api/Dockerfile new file mode 100644 index 0000000..f10e58e --- /dev/null +++ b/flask_api/Dockerfile @@ -0,0 +1,11 @@ + +FROM python:3.9-slim + +WORKDIR /app + +COPY requirements.txt requirements.txt +RUN pip install --no-cache-dir -r requirements.txt + +COPY . . + +CMD ["flask", "run", "--host=0.0.0.0", "--port=5000"] \ No newline at end of file diff --git a/flask_api/app.py b/flask_api/app.py new file mode 100644 index 0000000..9d95378 --- /dev/null +++ b/flask_api/app.py @@ -0,0 +1,207 @@ +from flask import Flask, request, jsonify +from flask_restful import Api, Resource +import psycopg2 +from psycopg2.extras import RealDictCursor + + +# Database connection +def get_db_connection(): + return psycopg2.connect( + host="postgres_container", + database="your_database", + user="your_username", + password="your_password", + cursor_factory=RealDictCursor + ) + + +def validate_server_state(state): + """Validate that server state is one of the allowed values.""" + valid_states = ['active', 'offline', 'retired'] + return state in valid_states if state else False + + +# Assuming we're using IPV4 +def validate_ip_address(ip_address): + """Basic validation for IP address format.""" + if not ip_address: + return False + parts = ip_address.split('.') + if len(parts) != 4: + return False + try: + return all(0 <= int(part) <= 255 for part in parts) + except ValueError: + return False + + +app = Flask(__name__) +api = Api(app) + + +class ServerList(Resource): + def get(self): + try: + with get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute("SELECT * FROM list_servers();") + servers = cur.fetchall() + return jsonify(servers) + except Exception as e: + # Do not return internal error details to client because this is a security risk. + print("Internal Server Error:", e) + return {"Internal Server Error."}, 500 + + + def post(self): + if not request.is_json: + return {"error": "Content-Type must be application/json"}, 400 + + data = request.get_json() + if not data: + return {"error": "Request body is required"}, 400 + + hostname = data.get("hostname") + ip_address = data.get("ip_address") + server_state = data.get("server_state") or data.get("state") + + # Validate required fields + if not hostname: + return {"error": "hostname is required"}, 400 + if not ip_address: + return {"error": "ip_address is required"}, 400 + if not server_state: + return {"error": "server_state is required"}, 400 + + # Validate server_state value + if not validate_server_state(server_state): + return {"error": "server_state must be one of: active, offline, retired"}, 400 + + # Validate IP address format + if not validate_ip_address(ip_address): + return {"error": "Invalid IP address format"}, 400 + + try: + with get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute( + "CALL insert_server(%s, %s, %s);", + (hostname, ip_address, server_state) + ) + conn.commit() + + # Get the inserted server ID + cur.execute( + "SELECT id FROM servers WHERE hostname = %s ORDER BY id DESC LIMIT 1;", + (hostname,) + ) + result = cur.fetchone() + if result: + server_id = result["id"] + else: + print ("Internal Server Error: Failed to retrieve new server ID") + return {"Internal Server Error"}, 500 + return {"server_id": server_id}, 201 + except psycopg2.IntegrityError as e: + if "unique" in str(e).lower() or "duplicate" in str(e).lower(): + return {"error": "Server with this hostname already exists"}, 400 + return {"error": "Invalid Input"}, 400 + except Exception as e: + print("Internal Server Error:", e) + return {"Internal Server Error."}, 500 + + +class Server(Resource): + def get(self, id): + try: + with get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute("SELECT * FROM get_server(%s);", (id,)) + server = cur.fetchone() + + if not server: + return {"error": "Server not found"}, 404 + return jsonify(server) + except Exception as e: + print("Internal Server Error:", e) + return {"Internal Server Error."}, 500 + + def put(self, id): + if not request.is_json: + return {"error": "Content-Type must be application/json"}, 400 + + data = request.get_json() + if not data: + return {"error": "Request body is required"}, 400 + + hostname = data.get("hostname") + ip_address = data.get("ip_address") + server_state = data.get("server_state") or data.get("state") + + # Validate that at least one field is provided + if not any([hostname, ip_address, server_state]): + return {"error": "At least one field (hostname, ip_address, server_state) must be provided"}, 400 + + # Validate server_state if provided + if server_state and not validate_server_state(server_state): + return {"error": "server_state must be one of: active, offline, retired"}, 400 + + # Validate IP address format if provided + if ip_address and not validate_ip_address(ip_address): + return {"error": "Invalid IP address format"}, 400 + + try: + with get_db_connection() as conn: + with conn.cursor() as cur: + # First check if server exists + cur.execute("SELECT * FROM get_server(%s);", (id,)) + existing_server = cur.fetchone() + + if not existing_server: + return {"error": "Server not found"}, 404 + + # Use provided values or existing values + update_hostname = hostname if hostname else existing_server["hostname"] + update_ip_address = ip_address if ip_address else existing_server["ip_address"] + update_server_state = server_state if server_state else existing_server["server_state"] + + # Call update stored procedure + cur.execute( + "CALL update_server(%s, %s, %s, %s);", + (id, update_hostname, update_ip_address, update_server_state) + ) + conn.commit() + + return {"id":id,"hostname":update_hostname,"ip_address":update_ip_address,"server_state":update_server_state}, 200 + except psycopg2.IntegrityError as e: + if "unique" in str(e).lower() or "duplicate" in str(e).lower(): + return {"error": "Server with this hostname already exists"}, 400 + return {"error": "Invalid Input"}, 400 + except Exception as e: + print("Internal Server Error:", e) + return {"Internal Server Error."}, 500 + + def delete(self, id): + try: + with get_db_connection() as conn: + with conn.cursor() as cur: + # First check if server exists + cur.execute("SELECT * FROM get_server(%s);", (id,)) + server = cur.fetchone() + + if not server: + return {"error": "Server not found"}, 404 + + cur.execute("CALL delete_server(%s);", (id,)) + conn.commit() + return {"message": "Server deleted"}, 200 + except Exception as e: + print("Internal Server Error:", e) + return {"Internal Server Error."}, 500 + + +api.add_resource(ServerList, '/servers') +api.add_resource(Server, '/servers/') + +if __name__ == '__main__': + app.run(debug=True) \ No newline at end of file diff --git a/flask_api/cli.py b/flask_api/cli.py new file mode 100755 index 0000000..8bb5549 --- /dev/null +++ b/flask_api/cli.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +""" +Command Line Interface for Server Inventory Management. + +This CLI provides commands to manage servers in the inventory via the REST API. +""" + +import click +import sys +import requests +from typing import Dict, Any + +# Import validation functions from app +from app import ( + validate_server_state, + validate_ip_address +) + + +# Configuration +API_BASE_URL = "http://localhost:5001" + + +def format_server(server: Dict[str, Any]) -> str: + """Format server data for display.""" + return f"ID: {server['id']}\n" \ + f"Hostname: {server['hostname']}\n" \ + f"IP Address: {server['ip_address']}\n" \ + f"State: {server['server_state']}" + + +@click.group() +@click.option('--api-url', default=API_BASE_URL, help='API base URL (default: http://localhost:5001)') +@click.pass_context +def cli(ctx, api_url): + """Server Inventory Management CLI.""" + ctx.ensure_object(dict) + ctx.obj['api_url'] = api_url + + +@cli.command() +@click.pass_context +def list(ctx): + """List all servers.""" + try: + response = requests.get(f"{ctx.obj['api_url']}/servers") + response.raise_for_status() + servers = response.json() + + if not servers: + click.echo("No servers found.") + return + + click.echo(f"\nFound {len(servers)} server(s):\n") + for server in servers: + click.echo(format_server(server)) + click.echo("-" * 40) + except requests.exceptions.RequestException as e: + click.echo(f"Error connecting to API: {e}", err=True) + sys.exit(1) + + +@cli.command() +@click.argument('server_id', type=int) +@click.pass_context +def get(ctx, server_id): + """Get a server by ID.""" + try: + response = requests.get(f"{ctx.obj['api_url']}/servers/{server_id}") + + if response.status_code == 404: + click.echo(f"Server with ID {server_id} not found.", err=True) + sys.exit(1) + + response.raise_for_status() + server = response.json() + click.echo("\n" + format_server(server)) + except requests.exceptions.RequestException as e: + click.echo(f"Error connecting to API: {e}", err=True) + sys.exit(1) + + +@cli.command() +@click.option('--hostname', required=True, help='Server hostname') +@click.option('--ip-address', required=True, help='Server IP address') +@click.option('--state', 'server_state', required=True, + type=click.Choice(['active', 'offline', 'retired'], case_sensitive=False), + help='Server state (active, offline, retired)') +@click.pass_context +def create(ctx, hostname, ip_address, server_state): + """Create a new server.""" + # Normalize state to lowercase + server_state = server_state.lower() + + # Validate inputs + if not validate_ip_address(ip_address): + click.echo("Error: Invalid IP address format.", err=True) + sys.exit(1) + + if not validate_server_state(server_state): + click.echo("Error: Invalid server state. Must be one of: active, offline, retired.", err=True) + sys.exit(1) + + try: + response = requests.post( + f"{ctx.obj['api_url']}/servers", + json={ + "hostname": hostname, + "ip_address": ip_address, + "server_state": server_state + }, + headers={"Content-Type": "application/json"} + ) + + if response.status_code == 400: + error_data = response.json() + click.echo(f"Error: {error_data.get('error', 'Bad request')}", err=True) + sys.exit(1) + + response.raise_for_status() + result = response.json() + click.echo(f"Server created successfully with ID: {result['server_id']}") + except requests.exceptions.RequestException as e: + click.echo(f"Error connecting to API: {e}", err=True) + sys.exit(1) + + +@cli.command() +@click.argument('server_id', type=int) +@click.option('--hostname', help='Update hostname') +@click.option('--ip-address', help='Update IP address') +@click.option('--state', 'server_state', + type=click.Choice(['active', 'offline', 'retired'], case_sensitive=False), + help='Update server state (active, offline, retired)') +@click.pass_context +def update(ctx, server_id, hostname, ip_address, server_state): + """Update a server by ID.""" + # Check that at least one field is provided + if not any([hostname, ip_address, server_state]): + click.echo("Error: At least one field (--hostname, --ip-address, or --state) must be provided.", err=True) + sys.exit(1) + + # Normalize state to lowercase if provided + if server_state: + server_state = server_state.lower() + + # Validate inputs if provided + if ip_address and not validate_ip_address(ip_address): + click.echo("Error: Invalid IP address format.", err=True) + sys.exit(1) + + if server_state and not validate_server_state(server_state): + click.echo("Error: Invalid server state. Must be one of: active, offline, retired.", err=True) + sys.exit(1) + + # Build update payload with only provided fields + payload = {} + if hostname: + payload['hostname'] = hostname + if ip_address: + payload['ip_address'] = ip_address + if server_state: + payload['server_state'] = server_state + + try: + response = requests.put( + f"{ctx.obj['api_url']}/servers/{server_id}", + json=payload, + headers={"Content-Type": "application/json"} + ) + + if response.status_code == 404: + click.echo(f"Error: Server with ID {server_id} not found.", err=True) + sys.exit(1) + + if response.status_code == 400: + error_data = response.json() + click.echo(f"Error: {error_data.get('error', 'Bad request')}", err=True) + sys.exit(1) + + response.raise_for_status() + result = response.json() + click.echo("Server updated successfully:") + click.echo(format_server(result)) + except requests.exceptions.RequestException as e: + click.echo(f"Error connecting to API: {e}", err=True) + sys.exit(1) + + +@cli.command() +@click.argument('server_id', type=int) +@click.confirmation_option(prompt='Are you sure you want to delete this server?') +@click.pass_context +def delete(ctx, server_id): + """Delete a server by ID.""" + try: + response = requests.delete(f"{ctx.obj['api_url']}/servers/{server_id}") + + if response.status_code == 404: + click.echo(f"Error: Server with ID {server_id} not found.", err=True) + sys.exit(1) + + response.raise_for_status() + result = response.json() + click.echo(result.get('message', 'Server deleted successfully.')) + except requests.exceptions.RequestException as e: + click.echo(f"Error connecting to API: {e}", err=True) + sys.exit(1) + + +if __name__ == '__main__': + cli() diff --git a/flask_api/requirements.txt b/flask_api/requirements.txt new file mode 100644 index 0000000..c8c9486 --- /dev/null +++ b/flask_api/requirements.txt @@ -0,0 +1,5 @@ +flask +flask-restful +psycopg2-binary +requests +click \ No newline at end of file diff --git a/flask_api/tests/__init__.py b/flask_api/tests/__init__.py new file mode 100644 index 0000000..e7991ee --- /dev/null +++ b/flask_api/tests/__init__.py @@ -0,0 +1,2 @@ +# Tests package + diff --git a/flask_api/tests/conftest.py b/flask_api/tests/conftest.py new file mode 100644 index 0000000..ffabbf6 --- /dev/null +++ b/flask_api/tests/conftest.py @@ -0,0 +1,15 @@ +""" +Pytest configuration file for tests. +This ensures that the flask_api directory is in the Python path +so that imports from app.py work correctly. +""" +import sys +from pathlib import Path + +# Add the parent directory (flask_api) to the Python path +# This allows imports like "from app import ..." to work +tests_dir = Path(__file__).parent +flask_api_dir = tests_dir.parent +if str(flask_api_dir) not in sys.path: + sys.path.insert(0, str(flask_api_dir)) + diff --git a/flask_api/tests/test_api_logic.py b/flask_api/tests/test_api_logic.py new file mode 100644 index 0000000..1a270af --- /dev/null +++ b/flask_api/tests/test_api_logic.py @@ -0,0 +1,409 @@ +import pytest +from unittest.mock import Mock, patch, MagicMock +import psycopg2 +from flask import Flask +from app import ServerList, Server + + +class TestCreateServerLogic: + """Unit tests for server creation logic.""" + + @patch('app.get_db_connection') + def test_create_server_returns_id(self, mock_db): + """Test that create server returns a server ID.""" + # Setup Flask app context + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + # First call returns the server ID, second call returns None (for the SELECT query) + mock_cursor.fetchone.side_effect = [{"id": 1}, {"id": 1}] + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers', method='POST', + json={"hostname": "test-server", "ip_address": "192.168.1.1", "server_state": "active"}): + server_list = ServerList() + response, status_code = server_list.post() + + assert status_code == 201 + assert response["server_id"] == 1 + mock_conn.commit.assert_called() + + @patch('app.get_db_connection') + def test_create_server_duplicate_hostname_error(self, mock_db): + """Test that duplicate hostname returns 400 error.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.execute.side_effect = psycopg2.IntegrityError("duplicate key value violates unique constraint") + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers', method='POST', + json={"hostname": "existing-server", "ip_address": "192.168.1.1", "server_state": "active"}): + server_list = ServerList() + response, status_code = server_list.post() + + assert status_code == 400 + assert "already exists" in response["error"] + + @patch('app.get_db_connection') + def test_create_server_commits_transaction(self, mock_db): + """Test that create server commits the transaction.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.side_effect = [{"id": 1}, {"id": 1}] + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers', method='POST', + json={"hostname": "test-server", "ip_address": "192.168.1.1", "server_state": "active"}): + server_list = ServerList() + server_list.post() + mock_conn.commit.assert_called_once() + + @patch('app.get_db_connection') + def test_create_server_missing_fields(self, mock_db): + """Test that missing required fields return 400.""" + app = Flask(__name__) + + with app.test_request_context('/servers', method='POST', json={"hostname": "test"}): + server_list = ServerList() + response, status_code = server_list.post() + + assert status_code == 400 + assert "required" in response["error"] + + @patch('app.get_db_connection') + def test_create_server_invalid_state(self, mock_db): + """Test that invalid server_state returns 400.""" + app = Flask(__name__) + + with app.test_request_context('/servers', method='POST', + json={"hostname": "test", "ip_address": "192.168.1.1", "server_state": "invalid"}): + server_list = ServerList() + response, status_code = server_list.post() + + assert status_code == 400 + assert "server_state must be one of" in response["error"] + + +class TestGetServerLogic: + """Unit tests for server retrieval logic.""" + + @patch('app.get_db_connection') + def test_get_server_returns_data(self, mock_db): + """Test that get server returns correct data.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = {"id": 1, "hostname": "test-server", "ip_address": "192.168.1.100", "server_state": "active"} + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers/1'): + server = Server() + response = server.get(1) + + # jsonify returns a Response object, so we need to get_json() + if hasattr(response, 'get_json'): + data = response.get_json() + else: + # If it's a tuple (error case), handle differently + data = response[0] if isinstance(response, tuple) else response + + assert data["id"] == 1 + assert data["hostname"] == "test-server" + assert data["ip_address"] == "192.168.1.100" + assert data["server_state"] == "active" + + @patch('app.get_db_connection') + def test_get_server_not_found(self, mock_db): + """Test that get server returns 404 when not found.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = None + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers/999'): + server = Server() + response, status_code = server.get(999) + + assert status_code == 404 + assert "not found" in response["error"] + + +class TestListServersLogic: + """Unit tests for server listing logic.""" + + @patch('app.get_db_connection') + def test_list_servers_returns_all(self, mock_db): + """Test that list servers returns all records.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchall.return_value = [ + {"id": 1, "hostname": "server-01", "ip_address": "192.168.1.1", "server_state": "active"}, + {"id": 2, "hostname": "server-02", "ip_address": "192.168.1.2", "server_state": "offline"}, + ] + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers'): + server_list = ServerList() + response = server_list.get() + + # jsonify returns a Response object, extract JSON data + if hasattr(response, 'get_json'): + data = response.get_json() + else: + # If it's a tuple (error case), handle differently + data = response[0] if isinstance(response, tuple) else response + + assert len(data) == 2 + assert data[0]["hostname"] == "server-01" + assert data[1]["hostname"] == "server-02" + + @patch('app.get_db_connection') + def test_list_servers_empty(self, mock_db): + """Test that list servers handles empty database.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchall.return_value = [] + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers'): + server_list = ServerList() + response = server_list.get() + + # jsonify returns a Response object, extract JSON data + if hasattr(response, 'get_json'): + data = response.get_json() + else: + data = response[0] if isinstance(response, tuple) else response + + assert len(data) == 0 + + +class TestUpdateServerLogic: + """Unit tests for server update logic.""" + + @patch('app.get_db_connection') + def test_update_server_single_field(self, mock_db): + """Test updating a single field.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = {"id": 1, "hostname": "old-name", "ip_address": "192.168.1.1", "server_state": "active"} + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers/1', method='PUT', + json={"hostname": "new-name"}): + server = Server() + response, status_code = server.put(1) + + assert status_code == 200 + assert response["hostname"] == "new-name" + mock_conn.commit.assert_called_once() + + @patch('app.get_db_connection') + def test_update_server_multiple_fields(self, mock_db): + """Test updating multiple fields.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = {"id": 1, "hostname": "old-name", "ip_address": "192.168.1.1", "server_state": "active"} + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers/1', method='PUT', + json={"hostname": "new-name", "ip_address": "10.0.0.1", "server_state": "offline"}): + server = Server() + response, status_code = server.put(1) + + assert status_code == 200 + assert response["hostname"] == "new-name" + assert response["ip_address"] == "10.0.0.1" + assert response["server_state"] == "offline" + + @patch('app.get_db_connection') + def test_update_server_not_found(self, mock_db): + """Test updating non-existent server returns 404.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = None + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers/999', method='PUT', + json={"hostname": "new-name"}): + server = Server() + response, status_code = server.put(999) + + assert status_code == 404 + assert "not found" in response["error"] + + @patch('app.get_db_connection') + def test_update_server_invalid_input(self, mock_db): + """Test that invalid input returns 400.""" + app = Flask(__name__) + + with app.test_request_context('/servers/1', method='PUT', + json={"server_state": "invalid"}): + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = {"id": 1, "hostname": "test", "ip_address": "192.168.1.1", "server_state": "active"} + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + server = Server() + response, status_code = server.put(1) + + assert status_code == 400 + assert "server_state must be one of" in response["error"] + + +class TestDeleteServerLogic: + """Unit tests for server deletion logic.""" + + @patch('app.get_db_connection') + def test_delete_server_success(self, mock_db): + """Test successful server deletion.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = {"id": 1, "hostname": "test-server", "ip_address": "192.168.1.1", "server_state": "active"} + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers/1', method='DELETE'): + server = Server() + response, status_code = server.delete(1) + + assert status_code == 200 + assert "deleted" in response["message"] + mock_conn.commit.assert_called_once() + + @patch('app.get_db_connection') + def test_delete_server_not_found(self, mock_db): + """Test deleting non-existent server returns 404.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = None + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers/999', method='DELETE'): + server = Server() + response, status_code = server.delete(999) + + assert status_code == 404 + assert "not found" in response["error"] + + +class TestDatabaseConnectionHandling: + """Unit tests for database connection handling.""" + + @patch('app.get_db_connection') + def test_connection_closes_on_error(self, mock_db): + """Test that connection closes properly on error.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.execute.side_effect = Exception("Database error") + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers'): + server_list = ServerList() + response, status_code = server_list.get() + + # Should return 500 error + assert status_code == 500 + # Connection context manager should handle cleanup + assert mock_conn.__exit__.called + + @patch('app.get_db_connection') + def test_transaction_handles_integrity_error(self, mock_db): + """Test that integrity errors are handled properly.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.execute.side_effect = psycopg2.IntegrityError("constraint violation") + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers', method='POST', + json={"hostname": "test", "ip_address": "192.168.1.1", "server_state": "active"}): + server_list = ServerList() + response, status_code = server_list.post() + + # Should return 400 for integrity error + assert status_code == 400 \ No newline at end of file diff --git a/flask_api/tests/test_cli.py b/flask_api/tests/test_cli.py new file mode 100644 index 0000000..a23d82c --- /dev/null +++ b/flask_api/tests/test_cli.py @@ -0,0 +1,635 @@ +""" +Integration tests for the Server Inventory Management CLI. + +These tests require the service to be running via docker-compose. +Run with: pytest flask_api/tests/test_cli.py -v + +Make sure docker-compose is running: + docker compose up -d +""" + +import pytest +import subprocess +import json +import time +import sys +from pathlib import Path + + +# CLI Configuration +CLI_SCRIPT = Path(__file__).parent.parent / "cli.py" +API_BASE_URL = "http://localhost:5001" +MAX_RETRIES = 30 +RETRY_DELAY = 1 + + +def run_cli_command(args, input_text=None, timeout=10): + """Run a CLI command and return the result.""" + cmd = [sys.executable, str(CLI_SCRIPT)] + args + try: + result = subprocess.run( + cmd, + input=input_text, + capture_output=True, + text=True, + timeout=timeout + ) + return { + "returncode": result.returncode, + "stdout": result.stdout, + "stderr": result.stderr, + "output": result.stdout + result.stderr + } + except subprocess.TimeoutExpired: + return { + "returncode": -1, + "stdout": "", + "stderr": "Command timed out", + "output": "Command timed out" + } + + +def wait_for_api(max_retries=MAX_RETRIES, delay=RETRY_DELAY): + """Wait for the API to be available.""" + import requests + for _ in range(max_retries): + try: + response = requests.get(f"{API_BASE_URL}/servers", timeout=2) + if response.status_code in [200, 404]: + return True + except Exception: + pass + time.sleep(delay) + return False + + +@pytest.fixture(scope="module", autouse=True) +def ensure_api_available(): + """Ensure API is available before running tests.""" + if not wait_for_api(): + pytest.skip("API is not available. Make sure docker-compose is running.") + + +@pytest.fixture(scope="function") +def cleanup_servers(): + """Cleanup fixture that removes all servers created during tests.""" + created_server_ids = [] + + yield created_server_ids + + # Cleanup: Delete all created servers via CLI + for server_id in created_server_ids: + try: + run_cli_command(["delete", str(server_id)], input_text="y\n", timeout=5) + except Exception: + pass # Ignore cleanup errors + + +def extract_server_id(output): + """Extract server ID from CLI output.""" + import re + match = re.search(r'ID:\s*(\d+)', output) + if match: + return int(match.group(1)) + # Try to find "with ID: X" pattern + match = re.search(r'with ID:\s*(\d+)', output) + if match: + return int(match.group(1)) + return None + + +class TestCLIList: + """Tests for CLI list command.""" + + def test_list_empty(self): + """Test listing servers when empty.""" + result = run_cli_command(["list"]) + assert result["returncode"] == 0 + assert "No servers found" in result["output"] or "Found 0 server" in result["output"] + + def test_list_with_servers(self, cleanup_servers): + """Test listing servers with data.""" + # Create servers via CLI + hostname1 = f"cli-list-test-1-{int(time.time())}" + hostname2 = f"cli-list-test-2-{int(time.time())}" + + result1 = run_cli_command([ + "create", + "--hostname", hostname1, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert result1["returncode"] == 0 + server_id1 = extract_server_id(result1["output"]) + if server_id1: + cleanup_servers.append(server_id1) + + result2 = run_cli_command([ + "create", + "--hostname", hostname2, + "--ip-address", "192.168.1.2", + "--state", "offline" + ]) + assert result2["returncode"] == 0 + server_id2 = extract_server_id(result2["output"]) + if server_id2: + cleanup_servers.append(server_id2) + + # List servers + result = run_cli_command(["list"]) + assert result["returncode"] == 0 + assert hostname1 in result["output"] + assert hostname2 in result["output"] + assert "192.168.1.1" in result["output"] + assert "192.168.1.2" in result["output"] + + +class TestCLICreate: + """Tests for CLI create command.""" + + def test_create_server_success(self, cleanup_servers): + """Test creating a server successfully.""" + hostname = f"cli-create-test-{int(time.time())}" + result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.100", + "--state", "active" + ]) + + assert result["returncode"] == 0 + assert "created successfully" in result["output"].lower() + assert "ID:" in result["output"] or "with ID:" in result["output"] + + server_id = extract_server_id(result["output"]) + if server_id: + cleanup_servers.append(server_id) + + def test_create_server_all_states(self, cleanup_servers): + """Test creating servers with all valid states.""" + valid_states = ["active", "offline", "retired"] + + for state in valid_states: + hostname = f"cli-state-{state}-{int(time.time())}" + result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "10.0.0.1", + "--state", state + ]) + + assert result["returncode"] == 0, f"Failed to create server with state {state}" + assert "created successfully" in result["output"].lower() + + server_id = extract_server_id(result["output"]) + if server_id: + cleanup_servers.append(server_id) + + def test_create_server_missing_hostname(self): + """Test creating server without hostname.""" + result = run_cli_command([ + "create", + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + + assert result["returncode"] != 0 + assert "hostname" in result["output"].lower() or "required" in result["output"].lower() + + def test_create_server_missing_ip(self): + """Test creating server without IP address.""" + result = run_cli_command([ + "create", + "--hostname", "test-server", + "--state", "active" + ]) + + assert result["returncode"] != 0 + assert "ip" in result["output"].lower() or "required" in result["output"].lower() + + def test_create_server_missing_state(self): + """Test creating server without state.""" + result = run_cli_command([ + "create", + "--hostname", "test-server", + "--ip-address", "192.168.1.1" + ]) + + assert result["returncode"] != 0 + assert "state" in result["output"].lower() or "required" in result["output"].lower() + + def test_create_server_invalid_ip(self): + """Test creating server with invalid IP.""" + result = run_cli_command([ + "create", + "--hostname", "test-server", + "--ip-address", "not.an.ip", + "--state", "active" + ]) + + assert result["returncode"] != 0 + assert "invalid" in result["output"].lower() or "ip" in result["output"].lower() + + def test_create_server_invalid_state(self): + """Test creating server with invalid state.""" + result = run_cli_command([ + "create", + "--hostname", "test-server", + "--ip-address", "192.168.1.1", + "--state", "invalid_state" + ]) + + # Click should prevent invalid choice before it reaches the API + assert result["returncode"] != 0 + + def test_create_server_duplicate_hostname(self, cleanup_servers): + """Test creating server with duplicate hostname.""" + hostname = f"cli-duplicate-{int(time.time())}" + + # Create first server + result1 = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert result1["returncode"] == 0 + server_id = extract_server_id(result1["output"]) + if server_id: + cleanup_servers.append(server_id) + + # Try to create duplicate + result2 = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.2", + "--state", "active" + ]) + + assert result2["returncode"] != 0 + assert "already exists" in result2["output"].lower() or "error" in result2["output"].lower() + + +class TestCLIGet: + """Tests for CLI get command.""" + + def test_get_server_success(self, cleanup_servers): + """Test getting a server by ID.""" + # Create a server first + hostname = f"cli-get-test-{int(time.time())}" + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.50", + "--state", "offline" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID from create output") + + cleanup_servers.append(server_id) + + # Get the server + result = run_cli_command(["get", str(server_id)]) + assert result["returncode"] == 0 + assert hostname in result["output"] + assert "192.168.1.50" in result["output"] + assert "offline" in result["output"] + assert f"ID: {server_id}" in result["output"] + + def test_get_server_not_found(self): + """Test getting a non-existent server.""" + result = run_cli_command(["get", "999999"]) + assert result["returncode"] != 0 + assert "not found" in result["output"].lower() + + def test_get_server_invalid_id(self): + """Test getting server with invalid ID format.""" + result = run_cli_command(["get", "invalid"]) + assert result["returncode"] != 0 + + +class TestCLIUpdate: + """Tests for CLI update command.""" + + def test_update_server_hostname(self, cleanup_servers): + """Test updating server hostname.""" + # Create a server + hostname = f"cli-update-{int(time.time())}" + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID") + + cleanup_servers.append(server_id) + + # Update hostname + new_hostname = "updated-hostname" + result = run_cli_command([ + "update", str(server_id), + "--hostname", new_hostname + ]) + + assert result["returncode"] == 0 + assert "updated successfully" in result["output"].lower() + assert new_hostname in result["output"] + + def test_update_server_ip(self, cleanup_servers): + """Test updating server IP address.""" + # Create a server + hostname = f"cli-update-ip-{int(time.time())}" + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID") + + cleanup_servers.append(server_id) + + # Update IP + result = run_cli_command([ + "update", str(server_id), + "--ip-address", "10.0.0.1" + ]) + + assert result["returncode"] == 0 + assert "10.0.0.1" in result["output"] + + def test_update_server_state(self, cleanup_servers): + """Test updating server state.""" + # Create a server + hostname = f"cli-update-state-{int(time.time())}" + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID") + + cleanup_servers.append(server_id) + + # Update state + result = run_cli_command([ + "update", str(server_id), + "--state", "offline" + ]) + + assert result["returncode"] == 0 + assert "offline" in result["output"] + + def test_update_server_multiple_fields(self, cleanup_servers): + """Test updating multiple fields.""" + # Create a server + hostname = f"cli-update-multi-{int(time.time())}" + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID") + + cleanup_servers.append(server_id) + + # Update multiple fields + result = run_cli_command([ + "update", str(server_id), + "--hostname", "multi-updated", + "--ip-address", "172.16.0.1", + "--state", "retired" + ]) + + assert result["returncode"] == 0 + assert "multi-updated" in result["output"] + assert "172.16.0.1" in result["output"] + assert "retired" in result["output"] + + def test_update_server_not_found(self): + """Test updating non-existent server.""" + result = run_cli_command([ + "update", "999999", + "--hostname", "new-name" + ]) + + assert result["returncode"] != 0 + assert "not found" in result["output"].lower() + + def test_update_server_no_fields(self, cleanup_servers): + """Test updating with no fields provided.""" + # Create a server + hostname = f"cli-update-no-fields-{int(time.time())}" + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID") + + cleanup_servers.append(server_id) + + # Try to update with no fields + result = run_cli_command(["update", str(server_id)]) + + assert result["returncode"] != 0 + assert "at least one field" in result["output"].lower() or "error" in result["output"].lower() + + def test_update_server_invalid_ip(self, cleanup_servers): + """Test updating with invalid IP.""" + # Create a server + hostname = f"cli-update-invalid-ip-{int(time.time())}" + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID") + + cleanup_servers.append(server_id) + + # Try to update with invalid IP + result = run_cli_command([ + "update", str(server_id), + "--ip-address", "invalid.ip" + ]) + + assert result["returncode"] != 0 + assert "invalid" in result["output"].lower() or "ip" in result["output"].lower() + + +class TestCLIDelete: + """Tests for CLI delete command.""" + + def test_delete_server_success(self): + """Test deleting a server.""" + # Create a server + hostname = f"cli-delete-test-{int(time.time())}" + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID") + + # Delete the server (with confirmation) + result = run_cli_command(["delete", str(server_id)], input_text="y\n") + + assert result["returncode"] == 0 + assert "deleted" in result["output"].lower() + + # Verify server is deleted + get_result = run_cli_command(["get", str(server_id)]) + assert get_result["returncode"] != 0 + assert "not found" in get_result["output"].lower() + + def test_delete_server_not_found(self): + """Test deleting non-existent server.""" + result = run_cli_command(["delete", "999999"], input_text="y\n") + assert result["returncode"] != 0 + assert "not found" in result["output"].lower() + + def test_delete_server_cancelled(self, cleanup_servers): + """Test cancelling delete operation.""" + # Create a server + hostname = f"cli-delete-cancel-{int(time.time())}" + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID") + + cleanup_servers.append(server_id) + + # Try to delete but cancel + result = run_cli_command(["delete", str(server_id)], input_text="n\n") + + # Should abort (non-zero exit code or specific message) + # The server should still exist + get_result = run_cli_command(["get", str(server_id)]) + assert get_result["returncode"] == 0 # Server should still exist + + +class TestCLIEndToEnd: + """End-to-end CLI workflow tests.""" + + def test_full_cli_workflow(self): + """Test a complete CLI workflow.""" + hostname = f"cli-e2e-{int(time.time())}" + + # Create + create_result = run_cli_command([ + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.200", + "--state", "active" + ]) + assert create_result["returncode"] == 0 + server_id = extract_server_id(create_result["output"]) + + if not server_id: + pytest.skip("Could not extract server ID") + + try: + # Read + get_result = run_cli_command(["get", str(server_id)]) + assert get_result["returncode"] == 0 + assert hostname in get_result["output"] + + # Update + update_result = run_cli_command([ + "update", str(server_id), + "--state", "offline" + ]) + assert update_result["returncode"] == 0 + + # Verify update + get_result2 = run_cli_command(["get", str(server_id)]) + assert get_result2["returncode"] == 0 + assert "offline" in get_result2["output"] + + # List (should include our server) + list_result = run_cli_command(["list"]) + assert list_result["returncode"] == 0 + assert hostname in list_result["output"] + + finally: + # Delete + delete_result = run_cli_command(["delete", str(server_id)], input_text="y\n") + assert delete_result["returncode"] == 0 + + +class TestCLIOptions: + """Tests for CLI options.""" + + def test_custom_api_url(self, cleanup_servers): + """Test using custom API URL.""" + hostname = f"cli-custom-url-{int(time.time())}" + result = run_cli_command([ + "--api-url", API_BASE_URL, + "create", + "--hostname", hostname, + "--ip-address", "192.168.1.1", + "--state", "active" + ]) + + assert result["returncode"] == 0 + server_id = extract_server_id(result["output"]) + if server_id: + cleanup_servers.append(server_id) + + def test_help_command(self): + """Test CLI help command.""" + result = run_cli_command(["--help"]) + assert result["returncode"] == 0 + assert "Server Inventory Management CLI" in result["output"] + + def test_list_help(self): + """Test list command help.""" + result = run_cli_command(["list", "--help"]) + assert result["returncode"] == 0 + assert "list" in result["output"].lower() + diff --git a/flask_api/tests/test_integration.py b/flask_api/tests/test_integration.py new file mode 100644 index 0000000..bb58907 --- /dev/null +++ b/flask_api/tests/test_integration.py @@ -0,0 +1,660 @@ +""" +Integration tests for the Server Inventory Management API. + +These tests require the service to be running via docker-compose. +Run with: pytest flask_api/tests/test_integration.py + +Make sure docker-compose is running: + docker compose up -d +""" + +import pytest +import requests +import time +from typing import Dict, Any, Optional + + +# API Configuration +API_BASE_URL = "http://localhost:5001" +API_TIMEOUT = 10 +MAX_RETRIES = 30 +RETRY_DELAY = 1 + + +def wait_for_api(max_retries: int = MAX_RETRIES, delay: int = RETRY_DELAY) -> bool: + """Wait for the API to be available.""" + for _ in range(max_retries): + try: + response = requests.get(f"{API_BASE_URL}/servers", timeout=2) + if response.status_code in [200, 404]: # 404 is OK, means API is up + return True + except requests.exceptions.RequestException: + pass + time.sleep(delay) + return False + + +@pytest.fixture(scope="module", autouse=True) +def ensure_api_available(): + """Ensure API is available before running tests.""" + if not wait_for_api(): + pytest.skip("API is not available. Make sure docker-compose is running.") + + +@pytest.fixture(scope="function") +def cleanup_servers(): + """Cleanup fixture that removes all servers created during tests.""" + created_server_ids = [] + + yield created_server_ids + + # Cleanup: Delete all created servers + for server_id in created_server_ids: + try: + requests.delete(f"{API_BASE_URL}/servers/{server_id}", timeout=API_TIMEOUT) + except requests.exceptions.RequestException: + pass # Ignore cleanup errors + + +class TestAPIConnectivity: + """Test basic API connectivity.""" + + def test_api_is_accessible(self): + """Test that the API is accessible.""" + response = requests.get(f"{API_BASE_URL}/servers", timeout=API_TIMEOUT) + assert response.status_code in [200, 404] # 200 or 404 (empty) is fine + + +class TestCreateServer: + """Integration tests for creating servers.""" + + def test_create_server_success(self, cleanup_servers): + """Test creating a server with valid data.""" + data = { + "hostname": f"test-server-{int(time.time())}", + "ip_address": "192.168.1.100", + "server_state": "active" + } + + response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 201 + result = response.json() + assert "server_id" in result + assert isinstance(result["server_id"], int) + cleanup_servers.append(result["server_id"]) + + def test_create_server_all_valid_states(self, cleanup_servers): + """Test creating servers with all valid states.""" + valid_states = ["active", "offline", "retired"] + + for state in valid_states: + data = { + "hostname": f"test-{state}-{int(time.time())}", + "ip_address": "10.0.0.1", + "server_state": state + } + + response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 201 + result = response.json() + cleanup_servers.append(result["server_id"]) + + def test_create_server_duplicate_hostname(self, cleanup_servers): + """Test that duplicate hostname returns 400.""" + hostname = f"duplicate-test-{int(time.time())}" + data = { + "hostname": hostname, + "ip_address": "192.168.1.1", + "server_state": "active" + } + + # Create first server + response1 = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + assert response1.status_code == 201 + cleanup_servers.append(response1.json()["server_id"]) + + # Try to create duplicate + response2 = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response2.status_code == 400 + error_data = response2.json() + assert "already exists" in error_data.get("error", "").lower() + + def test_create_server_missing_hostname(self): + """Test that missing hostname returns 400.""" + data = { + "ip_address": "192.168.1.1", + "server_state": "active" + } + + response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 400 + error_data = response.json() + assert "hostname" in error_data.get("error", "").lower() + + def test_create_server_missing_ip_address(self): + """Test that missing IP address returns 400.""" + data = { + "hostname": "test-server", + "server_state": "active" + } + + response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 400 + error_data = response.json() + assert "ip_address" in error_data.get("error", "").lower() + + def test_create_server_missing_state(self): + """Test that missing server_state returns 400.""" + data = { + "hostname": "test-server", + "ip_address": "192.168.1.1" + } + + response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 400 + error_data = response.json() + assert "server_state" in error_data.get("error", "").lower() + + def test_create_server_invalid_ip_format(self): + """Test that invalid IP format returns 400.""" + data = { + "hostname": "test-server", + "ip_address": "not.an.ip.address", + "server_state": "active" + } + + response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 400 + error_data = response.json() + assert "ip" in error_data.get("error", "").lower() + + def test_create_server_invalid_state(self): + """Test that invalid state returns 400.""" + data = { + "hostname": "test-server", + "ip_address": "192.168.1.1", + "server_state": "invalid_state" + } + + response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 400 + error_data = response.json() + assert "server_state" in error_data.get("error", "").lower() or "active, offline, retired" in error_data.get("error", "") + + +class TestListServers: + """Integration tests for listing servers.""" + + def test_list_servers_empty(self): + """Test listing servers when database is empty (or after cleanup).""" + response = requests.get(f"{API_BASE_URL}/servers", timeout=API_TIMEOUT) + + # Should return 200 even if empty + assert response.status_code == 200 + servers = response.json() + assert isinstance(servers, list) + + def test_list_servers_with_data(self, cleanup_servers): + """Test listing servers with data.""" + # Create a few servers + server_ids = [] + for i in range(3): + data = { + "hostname": f"list-test-{i}-{int(time.time())}", + "ip_address": f"192.168.1.{i+1}", + "server_state": "active" + } + response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + assert response.status_code == 201 + server_ids.append(response.json()["server_id"]) + cleanup_servers.extend(server_ids) + + # List servers + response = requests.get(f"{API_BASE_URL}/servers", timeout=API_TIMEOUT) + assert response.status_code == 200 + servers = response.json() + assert isinstance(servers, list) + assert len(servers) >= 3 + + # Verify all created servers are in the list + server_ids_in_response = [s["id"] for s in servers] + for server_id in server_ids: + assert server_id in server_ids_in_response + + +class TestGetServer: + """Integration tests for getting a single server.""" + + def test_get_server_success(self, cleanup_servers): + """Test getting a server by ID.""" + # Create a server + data = { + "hostname": f"get-test-{int(time.time())}", + "ip_address": "192.168.1.50", + "server_state": "offline" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + assert create_response.status_code == 201 + server_id = create_response.json()["server_id"] + cleanup_servers.append(server_id) + + # Get the server + response = requests.get(f"{API_BASE_URL}/servers/{server_id}", timeout=API_TIMEOUT) + assert response.status_code == 200 + server = response.json() + + assert server["id"] == server_id + assert server["hostname"] == data["hostname"] + assert server["ip_address"] == data["ip_address"] + assert server["server_state"] == data["server_state"] + + def test_get_server_not_found(self): + """Test getting a non-existent server returns 404.""" + # Use a very large ID that shouldn't exist + response = requests.get(f"{API_BASE_URL}/servers/999999", timeout=API_TIMEOUT) + assert response.status_code == 404 + error_data = response.json() + assert "not found" in error_data.get("error", "").lower() + + def test_get_server_invalid_id(self): + """Test getting a server with invalid ID format.""" + response = requests.get(f"{API_BASE_URL}/servers/invalid", timeout=API_TIMEOUT) + # Flask-RESTful should return 404 for invalid ID format + assert response.status_code == 404 + + +class TestUpdateServer: + """Integration tests for updating servers.""" + + def test_update_server_hostname(self, cleanup_servers): + """Test updating server hostname.""" + # Create a server + data = { + "hostname": f"update-test-{int(time.time())}", + "ip_address": "192.168.1.100", + "server_state": "active" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + server_id = create_response.json()["server_id"] + cleanup_servers.append(server_id) + + # Update hostname + update_data = {"hostname": "updated-hostname"} + response = requests.put( + f"{API_BASE_URL}/servers/{server_id}", + json=update_data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 200 + result = response.json() + assert result["hostname"] == "updated-hostname" + assert result["id"] == server_id + + def test_update_server_ip_address(self, cleanup_servers): + """Test updating server IP address.""" + # Create a server + data = { + "hostname": f"update-ip-{int(time.time())}", + "ip_address": "192.168.1.1", + "server_state": "active" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + server_id = create_response.json()["server_id"] + cleanup_servers.append(server_id) + + # Update IP address + update_data = {"ip_address": "10.0.0.1"} + response = requests.put( + f"{API_BASE_URL}/servers/{server_id}", + json=update_data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 200 + result = response.json() + assert result["ip_address"] == "10.0.0.1" + + def test_update_server_state(self, cleanup_servers): + """Test updating server state.""" + # Create a server + data = { + "hostname": f"update-state-{int(time.time())}", + "ip_address": "192.168.1.1", + "server_state": "active" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + server_id = create_response.json()["server_id"] + cleanup_servers.append(server_id) + + # Update state + update_data = {"server_state": "offline"} + response = requests.put( + f"{API_BASE_URL}/servers/{server_id}", + json=update_data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 200 + result = response.json() + assert result["server_state"] == "offline" + + def test_update_server_multiple_fields(self, cleanup_servers): + """Test updating multiple fields at once.""" + # Create a server + data = { + "hostname": f"update-multi-{int(time.time())}", + "ip_address": "192.168.1.1", + "server_state": "active" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + server_id = create_response.json()["server_id"] + cleanup_servers.append(server_id) + + # Update multiple fields + update_data = { + "hostname": "multi-updated", + "ip_address": "172.16.0.1", + "server_state": "retired" + } + response = requests.put( + f"{API_BASE_URL}/servers/{server_id}", + json=update_data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 200 + result = response.json() + assert result["hostname"] == "multi-updated" + assert result["ip_address"] == "172.16.0.1" + assert result["server_state"] == "retired" + + def test_update_server_not_found(self): + """Test updating non-existent server returns 404.""" + update_data = {"hostname": "new-name"} + response = requests.put( + f"{API_BASE_URL}/servers/999999", + json=update_data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 404 + error_data = response.json() + assert "not found" in error_data.get("error", "").lower() + + def test_update_server_invalid_ip(self, cleanup_servers): + """Test updating with invalid IP returns 400.""" + # Create a server + data = { + "hostname": f"invalid-ip-{int(time.time())}", + "ip_address": "192.168.1.1", + "server_state": "active" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + server_id = create_response.json()["server_id"] + cleanup_servers.append(server_id) + + # Try to update with invalid IP + update_data = {"ip_address": "invalid.ip"} + response = requests.put( + f"{API_BASE_URL}/servers/{server_id}", + json=update_data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 400 + error_data = response.json() + assert "ip" in error_data.get("error", "").lower() + + def test_update_server_invalid_state(self, cleanup_servers): + """Test updating with invalid state returns 400.""" + # Create a server + data = { + "hostname": f"invalid-state-{int(time.time())}", + "ip_address": "192.168.1.1", + "server_state": "active" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + server_id = create_response.json()["server_id"] + cleanup_servers.append(server_id) + + # Try to update with invalid state + update_data = {"server_state": "invalid_state"} + response = requests.put( + f"{API_BASE_URL}/servers/{server_id}", + json=update_data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 400 + error_data = response.json() + assert "server_state" in error_data.get("error", "").lower() + + def test_update_server_no_fields(self, cleanup_servers): + """Test updating with no fields returns 400.""" + # Create a server + data = { + "hostname": f"no-fields-{int(time.time())}", + "ip_address": "192.168.1.1", + "server_state": "active" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + server_id = create_response.json()["server_id"] + cleanup_servers.append(server_id) + + # Try to update with empty payload + response = requests.put( + f"{API_BASE_URL}/servers/{server_id}", + json={}, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + + assert response.status_code == 400 + error_data = response.json() + assert "request body is required" in error_data.get("error", "").lower() + + +class TestDeleteServer: + """Integration tests for deleting servers.""" + + def test_delete_server_success(self): + """Test deleting a server.""" + # Create a server + data = { + "hostname": f"delete-test-{int(time.time())}", + "ip_address": "192.168.1.1", + "server_state": "active" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + server_id = create_response.json()["server_id"] + + # Delete the server + response = requests.delete(f"{API_BASE_URL}/servers/{server_id}", timeout=API_TIMEOUT) + assert response.status_code == 200 + result = response.json() + assert "deleted" in result.get("message", "").lower() + + # Verify server is deleted + get_response = requests.get(f"{API_BASE_URL}/servers/{server_id}", timeout=API_TIMEOUT) + assert get_response.status_code == 404 + + def test_delete_server_not_found(self): + """Test deleting non-existent server returns 404.""" + response = requests.delete(f"{API_BASE_URL}/servers/999999", timeout=API_TIMEOUT) + assert response.status_code == 404 + error_data = response.json() + assert "not found" in error_data.get("error", "").lower() + + +class TestEndToEndWorkflow: + """End-to-end workflow tests.""" + + def test_full_crud_workflow(self): + """Test a complete CRUD workflow.""" + # Create + data = { + "hostname": f"e2e-test-{int(time.time())}", + "ip_address": "192.168.1.200", + "server_state": "active" + } + + create_response = requests.post( + f"{API_BASE_URL}/servers", + json=data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + assert create_response.status_code == 201 + server_id = create_response.json()["server_id"] + + try: + # Read + get_response = requests.get(f"{API_BASE_URL}/servers/{server_id}", timeout=API_TIMEOUT) + assert get_response.status_code == 200 + server = get_response.json() + assert server["hostname"] == data["hostname"] + + # Update + update_data = {"server_state": "offline"} + update_response = requests.put( + f"{API_BASE_URL}/servers/{server_id}", + json=update_data, + headers={"Content-Type": "application/json"}, + timeout=API_TIMEOUT + ) + assert update_response.status_code == 200 + updated_server = update_response.json() + assert updated_server["server_state"] == "offline" + + # Verify in list + list_response = requests.get(f"{API_BASE_URL}/servers", timeout=API_TIMEOUT) + assert list_response.status_code == 200 + servers = list_response.json() + server_ids = [s["id"] for s in servers] + assert server_id in server_ids + + finally: + # Delete + delete_response = requests.delete(f"{API_BASE_URL}/servers/{server_id}", timeout=API_TIMEOUT) + assert delete_response.status_code == 200 + diff --git a/flask_api/tests/test_validation.py b/flask_api/tests/test_validation.py new file mode 100644 index 0000000..dd6229e --- /dev/null +++ b/flask_api/tests/test_validation.py @@ -0,0 +1,183 @@ +import pytest +from unittest.mock import Mock, patch +from flask import Flask +from app import validate_ip_address, validate_server_state, ServerList, Server + + +class TestIPValidation: + """Unit tests for IP address validation using app validation function.""" + + @pytest.mark.parametrize("ip_address", [ + "192.168.1.1", + "10.0.0.1", + "172.16.0.1", + "0.0.0.0", + "255.255.255.255", + ]) + def test_valid_ip_addresses(self, ip_address): + """Test that valid IP addresses pass validation.""" + assert validate_ip_address(ip_address) is True, f"Valid IP {ip_address} failed validation" + + @pytest.mark.parametrize("ip_address", [ + "192.168.1", + "192.168.1.256", + "192.168.1.1.1", + "not.an.ip", + "192.168.-1.1", + "", + "192.168.1.1/24", + "192.168.1.1:8080", + None, + ]) + def test_invalid_ip_addresses(self, ip_address): + """Test that invalid IP addresses fail validation.""" + assert validate_ip_address(ip_address) is False, f"Invalid IP {ip_address} should have failed validation" + + @patch('app.get_db_connection') + def test_create_server_invalid_ip_returns_400(self, mock_db): + """Test that creating a server with invalid IP returns 400.""" + app = Flask(__name__) + + with app.test_request_context('/servers', method='POST', + json={"hostname": "test-server", "ip_address": "invalid.ip", "server_state": "active"}): + server_list = ServerList() + response, status_code = server_list.post() + + assert status_code == 400 + assert "Invalid IP address format" in response["error"] + + @patch('app.get_db_connection') + def test_create_server_missing_ip_returns_400(self, mock_db): + """Test that creating a server without IP returns 400.""" + app = Flask(__name__) + + with app.test_request_context('/servers', method='POST', + json={"hostname": "test-server", "server_state": "active"}): + server_list = ServerList() + response, status_code = server_list.post() + + assert status_code == 400 + assert "ip_address is required" in response["error"] + + @patch('app.get_db_connection') + def test_update_server_invalid_ip_returns_400(self, mock_db): + """Test that updating a server with invalid IP returns 400.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = {"id": 1, "hostname": "test", "ip_address": "192.168.1.1", "server_state": "active"} + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers/1', method='PUT', + json={"ip_address": "not.an.ip"}): + server = Server() + response, status_code = server.put(1) + + assert status_code == 400 + assert "Invalid IP address format" in response["error"] + + +class TestServerStateValidation: + """Unit tests for server state validation using app validation function.""" + + VALID_STATES = ["active", "offline", "retired"] + + @pytest.mark.parametrize("state", VALID_STATES) + def test_valid_states(self, state): + """Test that valid states pass validation.""" + assert validate_server_state(state) is True, f"Valid state {state} should pass validation" + + @pytest.mark.parametrize("state", [ + "ACTIVE", + "running", + "stopped", + "pending", + "", + None, + "active ", + " active", + "Active", + "OFFLINE", + ]) + def test_invalid_states(self, state): + """Test that invalid states fail validation.""" + assert validate_server_state(state) is False, f"Invalid state {state} should fail validation" + + @patch('app.get_db_connection') + def test_create_server_invalid_state_returns_400(self, mock_db): + """Test that creating a server with invalid state returns 400.""" + app = Flask(__name__) + + with app.test_request_context('/servers', method='POST', + json={"hostname": "test-server", "ip_address": "192.168.1.1", "server_state": "invalid"}): + server_list = ServerList() + response, status_code = server_list.post() + + assert status_code == 400 + assert "server_state must be one of" in response["error"] + + @patch('app.get_db_connection') + def test_create_server_missing_state_returns_400(self, mock_db): + """Test that creating a server without state returns 400.""" + app = Flask(__name__) + + with app.test_request_context('/servers', method='POST', + json={"hostname": "test-server", "ip_address": "192.168.1.1"}): + server_list = ServerList() + response, status_code = server_list.post() + + assert status_code == 400 + assert "server_state is required" in response["error"] + + @patch('app.get_db_connection') + def test_update_server_invalid_state_returns_400(self, mock_db): + """Test that updating a server with invalid state returns 400.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.return_value = {"id": 1, "hostname": "test", "ip_address": "192.168.1.1", "server_state": "active"} + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers/1', method='PUT', + json={"server_state": "invalid_state"}): + server = Server() + response, status_code = server.put(1) + + assert status_code == 400 + assert "server_state must be one of" in response["error"] + + @pytest.mark.parametrize("state", VALID_STATES) + @patch('app.get_db_connection') + def test_create_server_valid_states_accepted(self, mock_db, state): + """Test that all valid states are accepted when creating a server.""" + app = Flask(__name__) + + mock_conn = Mock() + mock_cursor = Mock() + mock_cursor.fetchone.side_effect = [{"id": 1}, {"id": 1}] + mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = Mock(return_value=False) + mock_conn.__enter__ = Mock(return_value=mock_conn) + mock_conn.__exit__ = Mock(return_value=False) + mock_db.return_value = mock_conn + + with app.test_request_context('/servers', method='POST', + json={"hostname": f"test-{state}", "ip_address": "192.168.1.1", "server_state": state}): + server_list = ServerList() + response, status_code = server_list.post() + + # Should succeed (201) or fail for other reasons (like duplicate), but not validation error + assert status_code in [201, 400] + if status_code == 400: + # If it's 400, it shouldn't be a validation error for state + assert "server_state must be one of" not in response.get("error", "") \ No newline at end of file diff --git a/init_db.sql b/init_db.sql new file mode 100644 index 0000000..26f2a4a --- /dev/null +++ b/init_db.sql @@ -0,0 +1,78 @@ +-- Database initialization script to create tables and procedures. +CREATE TABLE servers ( + id SERIAL PRIMARY KEY, + hostname VARCHAR(255) UNIQUE NOT NULL, + ip_address INET NOT NULL, + server_state VARCHAR(10) CHECK (server_state IN ('active', 'offline', 'retired')) NOT NULL +); + +-- Use functions and stored procedures to protect against SQL injection and improve performance. + +-- Insert a new server +CREATE PROCEDURE insert_server(_hostname VARCHAR(255), _ip_address INET, _server_state VARCHAR) +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO servers(hostname, ip_address, server_state) + VALUES(_hostname, _ip_address, _server_state); +END; +$$; + +-- List all servers +CREATE FUNCTION list_servers() +RETURNS TABLE(id INT, hostname VARCHAR(255), ip_address INET, server_state VARCHAR(10)) +LANGUAGE plpgsql +AS $$ +BEGIN + RETURN QUERY SELECT * FROM servers; +END; +$$; + +-- Update a server +CREATE PROCEDURE update_server(_id INT, _hostname VARCHAR(255), _ip_address INET, _server_state VARCHAR) +LANGUAGE plpgsql +AS $$ +BEGIN + UPDATE servers + SET hostname = _hostname, + ip_address = _ip_address, + server_state = _server_state + WHERE id = _id; +END; +$$; + +-- Get a server by ID +CREATE FUNCTION get_server(_id INT) +RETURNS TABLE(id INT, hostname VARCHAR(255), ip_address INET, server_state VARCHAR(10)) +LANGUAGE plpgsql +AS $$ +BEGIN + RETURN QUERY SELECT * FROM servers WHERE servers.id = _id; +END; +$$; + +-- Delete a server +CREATE PROCEDURE delete_server(_id INT) +LANGUAGE plpgsql +AS $$ +BEGIN + DELETE FROM servers + WHERE id = _id; +END; +$$; + + + + + + + + + + + + + + + + diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..333a8c8 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +pytest +flask +flask-restful +psycopg2-binary +requests +click \ No newline at end of file diff --git a/scripts/cli-test.sh b/scripts/cli-test.sh new file mode 100755 index 0000000..fb192ac --- /dev/null +++ b/scripts/cli-test.sh @@ -0,0 +1,52 @@ +#!/bin/bash +# CLI test script +# This script starts the docker-compose services and runs CLI tests + +set -e # Exit on error + +# Create a virtual environment and install dependencies +python3 -m venv challenge-env +source challenge-env/bin/activate +pip install -r requirements.txt + +# Make sure all services are stopped +docker compose down + +echo "Starting docker-compose services..." +docker compose up --build -d + +echo "Waiting for services to be ready..." +sleep 5 + +# Wait for API to be available +MAX_RETRIES=30 +RETRY_COUNT=0 +API_URL="http://localhost:5001" + +while [ $RETRY_COUNT -lt $MAX_RETRIES ]; do + if curl -s -f "$API_URL/servers" > /dev/null 2>&1; then + echo "API is ready!" + break + fi + RETRY_COUNT=$((RETRY_COUNT + 1)) + echo "Waiting for API... ($RETRY_COUNT/$MAX_RETRIES)" + sleep 2 +done + +if [ $RETRY_COUNT -eq $MAX_RETRIES ]; then + echo "ERROR: API did not become available in time" + docker compose logs flask_api + docker compose down + exit 1 +fi + +echo "Running CLI tests..." +pytest flask_api/tests/test_cli.py -v + +TEST_EXIT_CODE=$? + +echo "Stopping docker-compose services..." +docker compose down + +exit $TEST_EXIT_CODE + diff --git a/scripts/integration-test.sh b/scripts/integration-test.sh new file mode 100755 index 0000000..61e5e07 --- /dev/null +++ b/scripts/integration-test.sh @@ -0,0 +1,54 @@ +#!/bin/bash +# Integration test script +# This script starts the docker-compose services and runs integration tests + +set -e # Exit on error + +# Create a virtual environment and install dependencies +python3 -m venv challenge-env +source challenge-env/bin/activate +pip install -r requirements.txt + +# Make sure all services are stopped +docker compose down + +echo "Starting docker-compose services..." +docker compose up --build -d + +echo "Waiting for services to be ready..." +sleep 5 + +# Wait for API to be available +MAX_RETRIES=30 +RETRY_COUNT=0 +API_URL="http://localhost:5001" + +while [ $RETRY_COUNT -lt $MAX_RETRIES ]; do + if curl -s -f "$API_URL/servers" > /dev/null 2>&1; then + echo "API is ready!" + break + fi + RETRY_COUNT=$((RETRY_COUNT + 1)) + echo "Waiting for API... ($RETRY_COUNT/$MAX_RETRIES)" + sleep 2 +done + +if [ $RETRY_COUNT -eq $MAX_RETRIES ]; then + echo "ERROR: API did not become available in time" + docker compose logs flask_api + docker compose down + exit 1 +fi + +echo "Running integration tests..." +pytest flask_api/tests/test_integration.py -v + +TEST_EXIT_CODE=$? + +echo "Stopping docker-compose services..." +docker compose down + +# Deactivate virtual environment +deactivate + +exit $TEST_EXIT_CODE diff --git a/scripts/unit-test.sh b/scripts/unit-test.sh new file mode 100755 index 0000000..658334a --- /dev/null +++ b/scripts/unit-test.sh @@ -0,0 +1,15 @@ +#! /bin/bash + +set -e # Exit on error + +# Create a virtual environment and install dependencies +python3 -m venv challenge-env +source challenge-env/bin/activate +pip install -r requirements.txt + +# Run unit tests +pytest flask_api/tests/test_api_logic.py -v +pytest flask_api/tests/test_validation.py -v + +# Deactivate virtual environment +deactivate \ No newline at end of file