105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
import sys
|
|
import os
|
|
import pytest
|
|
from flask import Flask
|
|
from flask_httpauth import HTTPBasicAuth
|
|
import base64
|
|
from redis import from_url, ConnectionPool, Redis
|
|
from flask_limiter import Limiter
|
|
from flask_limiter.util import get_remote_address
|
|
from unittest.mock import patch, Mock
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../app')))
|
|
from routes import setup_routes
|
|
|
|
auth = HTTPBasicAuth()
|
|
|
|
# Force configuration to use fallback values
|
|
|
|
class TestConfig():
|
|
SECRET_KEY = '28c93e98b2a87e47db16372bdb6e7593fb1addf9ccc10eae562827a7358cab3b'
|
|
|
|
REDIS_URL="redis://localhost:6379/0"
|
|
RATE_LIMITS = ["200 per day", "50 per hour"]
|
|
|
|
def setup_redis(app):
|
|
pool = ConnectionPool.from_url(REDIS_URL)
|
|
redis_connection = Redis(connection_pool=pool)
|
|
redis_connection.ping()
|
|
app.logger.info("Connected to Redis")
|
|
# Return the redis_connection to be used elsewhere
|
|
return redis_connection
|
|
|
|
def setup_limiter(app):
|
|
|
|
# Rate limiter setup
|
|
limiter = Limiter(
|
|
key_func=get_remote_address,
|
|
app=app,
|
|
storage_uri=REDIS_URL,
|
|
default_limits=RATE_LIMITS
|
|
)
|
|
|
|
return limiter
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
app = Flask(__name__)
|
|
redis_connection = setup_redis(app)
|
|
app.config.from_object(TestConfig)
|
|
setup_routes(setup_limiter(app), app, redis_connection)
|
|
return app
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
return app.test_client()
|
|
|
|
@pytest.fixture
|
|
def auth_headers():
|
|
password = '4acG2wHmp1-B'
|
|
credentials = f"anon:{password}"
|
|
encoded_credentials = base64.b64encode(credentials.encode()).decode('utf-8')
|
|
headers = {
|
|
'Authorization': f'Basic {encoded_credentials}'
|
|
}
|
|
return headers
|
|
|
|
def mock_subprocess_run(*args, **kwargs):
|
|
"""Mock function to simulate subprocess.run behavior."""
|
|
class MockCompletedProcess:
|
|
def __init__(self, returncode=0, stdout='', stderr=''):
|
|
self.returncode = returncode
|
|
self.stdout = stdout
|
|
self.stderr = stderr
|
|
|
|
if 'ping' in args[0]:
|
|
if '192.168.0.0' in args[0]:
|
|
# Simulate unreachable IP
|
|
return MockCompletedProcess(returncode=1, stderr='Unreachable')
|
|
else:
|
|
# Simulate reachable IP
|
|
return MockCompletedProcess(returncode=0, stdout='Ping successful')
|
|
return MockCompletedProcess()
|
|
|
|
def test_wol_valid_device(client, auth_headers):
|
|
with patch.dict(os.environ, {'LAPTOP_MAC': '00:00:00:00:00:00', 'LAPTOP_IP': '192.168.0.1'}):
|
|
with patch('subprocess.run', mock_subprocess_run):
|
|
response = client.get('/wol/laptop?key=28c93e98b2a87e47db16372bdb6e7593fb1addf9ccc10eae562827a7358cab3b', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
def test_unauthorized_wol(client):
|
|
response = client.get('/wol/device1?key=wrongkey')
|
|
assert response.status_code == 401
|
|
|
|
def test_invalid_device(client, auth_headers):
|
|
response = client.get('/wol/invalid_device?key=28c93e98b2a87e47db16372bdb6e7593fb1addf9ccc10eae562827a7358cab3b', headers=auth_headers)
|
|
assert response.status_code == 400
|
|
|
|
def test_redis_status(client, auth_headers):
|
|
response = client.get('/redis_status', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
def test_reset_limiter(client, auth_headers):
|
|
response = client.get('/reset_limiter?key=28c93e98b2a87e47db16372bdb6e7593fb1addf9ccc10eae562827a7358cab3b', headers=auth_headers)
|
|
assert response.status_code == 200
|