555 lines
21 KiB
Python
555 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Shaarli API Complete Test Script
|
|
Tests all available endpoints of the Shaarli REST API
|
|
"""
|
|
|
|
import jwt
|
|
import time
|
|
import requests
|
|
import json
|
|
import sys
|
|
import io
|
|
from datetime import datetime
|
|
from typing import Optional, Dict, Any, List
|
|
|
|
# Fix Windows console encoding
|
|
if sys.platform == 'win32':
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
|
|
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
|
|
|
|
# --- CONFIGURATION ---
|
|
SHAARLI_URL = "https://bm.dracodev.net"
|
|
API_SECRET = "Chab30017405"
|
|
# ---------------------
|
|
|
|
class ShaarliAPITester:
|
|
"""Complete Shaarli API Tester"""
|
|
|
|
def __init__(self, base_url: str, api_secret: str):
|
|
self.base_url = base_url.rstrip('/')
|
|
self.api_secret = api_secret
|
|
self.session = requests.Session()
|
|
self.test_link_id: Optional[int] = None
|
|
|
|
def generate_token(self) -> str:
|
|
"""Generate JWT HS512 token required by Shaarli"""
|
|
# iat must be slightly in the past to avoid clock skew issues
|
|
payload = {'iat': int(time.time()) - 60}
|
|
token = jwt.encode(payload, self.api_secret, algorithm='HS512')
|
|
if isinstance(token, bytes):
|
|
token = token.decode('utf-8')
|
|
return token
|
|
|
|
def get_headers(self) -> Dict[str, str]:
|
|
"""Get request headers with fresh JWT token"""
|
|
return {
|
|
'Authorization': f'Bearer {self.generate_token()}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
def api_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> requests.Response:
|
|
"""Make an API request"""
|
|
url = f"{self.base_url}/api/v1{endpoint}"
|
|
headers = self.get_headers()
|
|
|
|
if method.upper() == 'GET':
|
|
return self.session.get(url, headers=headers, params=data)
|
|
elif method.upper() == 'POST':
|
|
return self.session.post(url, headers=headers, json=data)
|
|
elif method.upper() == 'PUT':
|
|
return self.session.put(url, headers=headers, json=data)
|
|
elif method.upper() == 'DELETE':
|
|
return self.session.delete(url, headers=headers)
|
|
else:
|
|
raise ValueError(f"Unknown method: {method}")
|
|
|
|
def print_header(self, title: str):
|
|
"""Print a formatted section header"""
|
|
print(f"\n{'='*60}")
|
|
print(f" {title}")
|
|
print(f"{'='*60}")
|
|
|
|
def print_subheader(self, title: str):
|
|
"""Print a formatted subsection header"""
|
|
print(f"\n--- {title} ---")
|
|
|
|
def print_success(self, message: str):
|
|
print(f"[OK] {message}")
|
|
|
|
def print_error(self, message: str):
|
|
print(f"[FAIL] {message}")
|
|
|
|
def print_info(self, key: str, value: Any):
|
|
print(f" > {key}: {value}")
|
|
|
|
def print_item(self, message: str):
|
|
print(f" * {message}")
|
|
|
|
# ==================== INFO ENDPOINT ====================
|
|
|
|
def test_info(self) -> bool:
|
|
"""Test GET /info - Instance information"""
|
|
self.print_subheader("GET /info - Instance Information")
|
|
|
|
try:
|
|
response = self.api_request('GET', '/info')
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.print_success("Instance information retrieved!")
|
|
self.print_info("Global Links Count", data.get('global_counter', 'N/A'))
|
|
self.print_info("Private Links Count", data.get('private_counter', 'N/A'))
|
|
|
|
settings = data.get('settings', {})
|
|
if settings:
|
|
self.print_info("Title", settings.get('title', 'N/A'))
|
|
self.print_info("Header Link", settings.get('header_link', 'N/A'))
|
|
self.print_info("Timezone", settings.get('timezone', 'N/A'))
|
|
self.print_info("Default Private", settings.get('default_private_links', 'N/A'))
|
|
plugins = settings.get('enabled_plugins', [])
|
|
self.print_info("Enabled Plugins", ', '.join(plugins) if plugins else 'None')
|
|
return True
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}: {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
# ==================== LINKS ENDPOINTS ====================
|
|
|
|
def test_get_links(self, limit: int = 5) -> bool:
|
|
"""Test GET /links - Get links list"""
|
|
self.print_subheader(f"GET /links - Get Links (limit={limit})")
|
|
|
|
try:
|
|
response = self.api_request('GET', '/links', {'limit': limit})
|
|
|
|
if response.status_code == 200:
|
|
links = response.json()
|
|
self.print_success(f"Retrieved {len(links)} links")
|
|
|
|
for link in links[:5]: # Show max 5
|
|
title = link.get('title', 'No title')[:50]
|
|
url = link.get('url', '')[:40]
|
|
tags = ', '.join(link.get('tags', [])) or 'No tags'
|
|
private = "[PRIVATE]" if link.get('private') else "[PUBLIC]"
|
|
self.print_item(f"{private} [{link.get('id')}] {title}")
|
|
print(f" URL: {url}...")
|
|
print(f" Tags: {tags}")
|
|
return True
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}: {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
def test_get_links_with_search(self, searchterm: str) -> bool:
|
|
"""Test GET /links with search parameter"""
|
|
self.print_subheader(f"GET /links?searchterm={searchterm}")
|
|
|
|
try:
|
|
response = self.api_request('GET', '/links', {'searchterm': searchterm, 'limit': 5})
|
|
|
|
if response.status_code == 200:
|
|
links = response.json()
|
|
self.print_success(f"Found {len(links)} links matching '{searchterm}'")
|
|
for link in links[:3]:
|
|
self.print_item(f"[{link.get('id')}] {link.get('title', 'No title')[:50]}")
|
|
return True
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
def test_get_links_visibility(self, visibility: str) -> bool:
|
|
"""Test GET /links with visibility filter"""
|
|
self.print_subheader(f"GET /links?visibility={visibility}")
|
|
|
|
try:
|
|
response = self.api_request('GET', '/links', {'visibility': visibility, 'limit': 5})
|
|
|
|
if response.status_code == 200:
|
|
links = response.json()
|
|
self.print_success(f"Found {len(links)} {visibility} links")
|
|
return True
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
def test_create_link(self) -> bool:
|
|
"""Test POST /links - Create a new link"""
|
|
self.print_subheader("POST /links - Create New Link")
|
|
|
|
test_link = {
|
|
'url': f'https://example.com/test-{int(time.time())}',
|
|
'title': f'Test Link from API - {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}',
|
|
'description': 'This is a test link created by the Shaarli API tester script',
|
|
'tags': ['test', 'api', 'automated'],
|
|
'private': True
|
|
}
|
|
|
|
try:
|
|
response = self.api_request('POST', '/links', test_link)
|
|
|
|
if response.status_code in [200, 201]:
|
|
link = response.json()
|
|
self.test_link_id = link.get('id')
|
|
self.print_success(f"Link created successfully!")
|
|
self.print_info("ID", link.get('id'))
|
|
self.print_info("Title", link.get('title'))
|
|
self.print_info("URL", link.get('url'))
|
|
self.print_info("Short URL", link.get('shorturl'))
|
|
self.print_info("Private", link.get('private'))
|
|
self.print_info("Tags", ', '.join(link.get('tags', [])))
|
|
return True
|
|
elif response.status_code == 409:
|
|
self.print_error("Conflict - Link URL already exists")
|
|
return False
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}: {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
def test_get_single_link(self, link_id: int) -> bool:
|
|
"""Test GET /links/{id} - Get a specific link"""
|
|
self.print_subheader(f"GET /links/{link_id} - Get Single Link")
|
|
|
|
try:
|
|
response = self.api_request('GET', f'/links/{link_id}')
|
|
|
|
if response.status_code == 200:
|
|
link = response.json()
|
|
self.print_success("Link retrieved successfully!")
|
|
self.print_info("ID", link.get('id'))
|
|
self.print_info("Title", link.get('title'))
|
|
self.print_info("URL", link.get('url'))
|
|
self.print_info("Created", link.get('created'))
|
|
self.print_info("Updated", link.get('updated'))
|
|
return True
|
|
elif response.status_code == 404:
|
|
self.print_error(f"Link {link_id} not found")
|
|
return False
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
def test_update_link(self, link_id: int) -> bool:
|
|
"""Test PUT /links/{id} - Update a link"""
|
|
self.print_subheader(f"PUT /links/{link_id} - Update Link")
|
|
|
|
# First get the current link
|
|
try:
|
|
response = self.api_request('GET', f'/links/{link_id}')
|
|
if response.status_code != 200:
|
|
self.print_error(f"Cannot get link to update: {response.status_code}")
|
|
return False
|
|
|
|
current_link = response.json()
|
|
|
|
# Update the link
|
|
updated_data = {
|
|
'url': current_link.get('url'),
|
|
'title': f"{current_link.get('title')} [UPDATED]",
|
|
'description': f"{current_link.get('description', '')} - Updated at {datetime.now().strftime('%H:%M:%S')}",
|
|
'tags': current_link.get('tags', []) + ['updated'],
|
|
'private': current_link.get('private', True)
|
|
}
|
|
|
|
response = self.api_request('PUT', f'/links/{link_id}', updated_data)
|
|
|
|
if response.status_code == 200:
|
|
link = response.json()
|
|
self.print_success("Link updated successfully!")
|
|
self.print_info("New Title", link.get('title'))
|
|
self.print_info("New Tags", ', '.join(link.get('tags', [])))
|
|
return True
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}: {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
def test_delete_link(self, link_id: int) -> bool:
|
|
"""Test DELETE /links/{id} - Delete a link"""
|
|
self.print_subheader(f"DELETE /links/{link_id} - Delete Link")
|
|
|
|
try:
|
|
response = self.api_request('DELETE', f'/links/{link_id}')
|
|
|
|
if response.status_code in [200, 204]:
|
|
self.print_success(f"Link {link_id} deleted successfully!")
|
|
return True
|
|
elif response.status_code == 404:
|
|
self.print_error(f"Link {link_id} not found")
|
|
return False
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}: {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
# ==================== TAGS ENDPOINTS ====================
|
|
|
|
def test_get_tags(self, limit: int = 20) -> bool:
|
|
"""Test GET /tags - Get all tags"""
|
|
self.print_subheader(f"GET /tags - Get Tags (limit={limit})")
|
|
|
|
try:
|
|
response = self.api_request('GET', '/tags', {'limit': limit})
|
|
|
|
if response.status_code == 200:
|
|
tags = response.json()
|
|
self.print_success(f"Retrieved {len(tags)} tags")
|
|
|
|
# Show tags sorted by occurrences
|
|
for tag in tags[:15]:
|
|
name = tag.get('name', 'Unknown')
|
|
occurrences = tag.get('occurrences', 0)
|
|
bar = '#' * min(occurrences, 20)
|
|
self.print_item(f"{name}: {occurrences} {bar}")
|
|
|
|
if len(tags) > 15:
|
|
print(f" ... and {len(tags) - 15} more tags")
|
|
return True
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}: {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
def test_get_single_tag(self, tag_name: str) -> bool:
|
|
"""Test GET /tags/{tagName} - Get a specific tag"""
|
|
self.print_subheader(f"GET /tags/{tag_name}")
|
|
|
|
try:
|
|
response = self.api_request('GET', f'/tags/{tag_name}')
|
|
|
|
if response.status_code == 200:
|
|
tag = response.json()
|
|
self.print_success("Tag retrieved!")
|
|
self.print_info("Name", tag.get('name'))
|
|
self.print_info("Occurrences", tag.get('occurrences'))
|
|
return True
|
|
elif response.status_code == 404:
|
|
self.print_error(f"Tag '{tag_name}' not found")
|
|
return False
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
# ==================== HISTORY ENDPOINT ====================
|
|
|
|
def test_get_history(self, limit: int = 10) -> bool:
|
|
"""Test GET /history - Get recent actions"""
|
|
self.print_subheader(f"GET /history - Recent Actions (limit={limit})")
|
|
|
|
try:
|
|
response = self.api_request('GET', '/history', {'limit': limit})
|
|
|
|
if response.status_code == 200:
|
|
history = response.json()
|
|
self.print_success(f"Retrieved {len(history)} history entries")
|
|
|
|
action_icons = {
|
|
'CREATED': '[+]',
|
|
'UPDATED': '[~]',
|
|
'DELETED': '[-]',
|
|
'SETTINGS': '[=]'
|
|
}
|
|
|
|
for entry in history[:10]:
|
|
action = entry.get('event', 'UNKNOWN')
|
|
icon = action_icons.get(action, '[?]')
|
|
link_id = entry.get('id', 'N/A')
|
|
datetime_str = entry.get('datetime', 'Unknown time')
|
|
self.print_item(f"{icon} {action} - Link #{link_id} at {datetime_str}")
|
|
return True
|
|
else:
|
|
self.print_error(f"Failed with status {response.status_code}: {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Error: {e}")
|
|
return False
|
|
|
|
# ==================== MAIN TEST RUNNER ====================
|
|
|
|
def run_all_tests(self, include_write_tests: bool = True):
|
|
"""Run all API tests"""
|
|
print(f"\n{'#'*60}")
|
|
print(f"# SHAARLI API COMPLETE TEST SUITE")
|
|
print(f"# Target: {self.base_url}")
|
|
print(f"# Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print(f"{'#'*60}")
|
|
|
|
results = {
|
|
'passed': 0,
|
|
'failed': 0,
|
|
'skipped': 0
|
|
}
|
|
|
|
# ===== INFO =====
|
|
self.print_header("INSTANCE INFORMATION")
|
|
if self.test_info():
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
|
|
# ===== LINKS (READ) =====
|
|
self.print_header("LINKS - READ OPERATIONS")
|
|
|
|
if self.test_get_links(limit=5):
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
|
|
# Get first link ID for single link test
|
|
try:
|
|
response = self.api_request('GET', '/links', {'limit': 1})
|
|
if response.status_code == 200:
|
|
links = response.json()
|
|
if links:
|
|
first_link_id = links[0].get('id')
|
|
if self.test_get_single_link(first_link_id):
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
except:
|
|
results['skipped'] += 1
|
|
|
|
if self.test_get_links_visibility('public'):
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
|
|
if self.test_get_links_visibility('private'):
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
|
|
if self.test_get_links_with_search('test'):
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
|
|
# ===== TAGS =====
|
|
self.print_header("TAGS")
|
|
|
|
if self.test_get_tags():
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
|
|
# Get first tag for single tag test
|
|
try:
|
|
response = self.api_request('GET', '/tags', {'limit': 1})
|
|
if response.status_code == 200:
|
|
tags = response.json()
|
|
if tags:
|
|
first_tag = tags[0].get('name')
|
|
if self.test_get_single_tag(first_tag):
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
except:
|
|
results['skipped'] += 1
|
|
|
|
# ===== HISTORY =====
|
|
self.print_header("HISTORY")
|
|
|
|
if self.test_get_history():
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
|
|
# ===== LINKS (WRITE) =====
|
|
if include_write_tests:
|
|
self.print_header("LINKS - WRITE OPERATIONS (CREATE/UPDATE/DELETE)")
|
|
print("WARNING: These tests will create, modify, and delete a test link")
|
|
|
|
# Create
|
|
if self.test_create_link():
|
|
results['passed'] += 1
|
|
|
|
if self.test_link_id:
|
|
# Update
|
|
if self.test_update_link(self.test_link_id):
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
|
|
# Delete
|
|
if self.test_delete_link(self.test_link_id):
|
|
results['passed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
results['skipped'] += 2
|
|
else:
|
|
self.print_header("WRITE OPERATIONS - SKIPPED")
|
|
print(" (Use include_write_tests=True to test CREATE/UPDATE/DELETE)")
|
|
results['skipped'] += 3
|
|
|
|
# ===== SUMMARY =====
|
|
self.print_header("TEST SUMMARY")
|
|
total = results['passed'] + results['failed'] + results['skipped']
|
|
print(f"\n [OK] Passed: {results['passed']}")
|
|
print(f" [FAIL] Failed: {results['failed']}")
|
|
print(f" [SKIP] Skipped: {results['skipped']}")
|
|
print(f" ----------------")
|
|
print(f" Total: {total}")
|
|
|
|
if results['failed'] == 0:
|
|
print(f"\n SUCCESS! All tests passed! Your Shaarli API is working correctly.")
|
|
else:
|
|
print(f"\n WARNING: Some tests failed. Check the output above for details.")
|
|
|
|
return results
|
|
|
|
|
|
def main():
|
|
print("Starting Shaarli API Test Suite...")
|
|
|
|
tester = ShaarliAPITester(SHAARLI_URL, API_SECRET)
|
|
|
|
# Run all tests including write operations
|
|
# Set to False if you don't want to create/modify/delete test links
|
|
results = tester.run_all_tests(include_write_tests=True)
|
|
|
|
# Exit with error code if any tests failed
|
|
sys.exit(0 if results['failed'] == 0 else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |