mirror/maigret
1
0
Fork 0
mirror of https://github.com/soxoj/maigret.git synced 2026-05-06 14:08:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

CLI test fixes

Browse Source
This commit is contained in:
Soxoj
2024-12-15 12:57:01 +01:00
parent f8f7c996ca
commit 86ea0b9212
4 changed files with 119 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+1 -1
View File
@@ -1,7 +1,7 @@
LINT_FILES=maigret wizard.py tests
test:
coverage run --source=./maigret -m pytest tests
coverage run --source=./maigret,./maigret/web -m pytest tests
coverage report -m
coverage html
maigret/maigret.py
+5 -5
View File
@@ -330,11 +330,9 @@ def setup_arguments_parser(settings: Settings):
type=int,
nargs='?',
const=5000, # default if --web is provided without a port
default=None,
default=settings.web_interface_port,
help="Launches the web interface on the specified port (default: 5000 if no PORT is provided).",
)
output_group = parser.add_argument_group(
'Output options', 'Options to change verbosity and view of the console output'
)
@@ -497,9 +495,11 @@ async def main():
if args.web is not None:
from maigret.web.app import app
port = args.web if args.web else 5000 # args.web is either the specified port or 5000 by const
app.run(port=port)
port = (
args.web if args.web else 5000
) # args.web is either the specified port or 5000 by const
app.run(port=port)
# Usernames initial list
usernames = {
maigret/web/app.py
+79 -33
View File
@@ -1,5 +1,14 @@
# app.py
from flask import Flask, render_template, request, send_file, Response, flash, redirect, url_for
from flask import (
Flask,
render_template,
request,
send_file,
Response,
flash,
redirect,
url_for,
)
import logging
import os
import asyncio
@@ -26,11 +35,13 @@ REPORTS_FOLDER = os.path.abspath('/tmp/maigret_reports')
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
os.makedirs(REPORTS_FOLDER, exist_ok=True)
def setup_logger(log_level, name):
logger = logging.getLogger(name)
logger.setLevel(log_level)
return logger
async def maigret_search(username, options):
logger = setup_logger(logging.WARNING, 'maigret')
try:
@@ -50,6 +61,7 @@ async def maigret_search(username, options):
logger.error(f"Error during search: {str(e)}")
raise
async def search_multiple_usernames(usernames, options):
results = []
for username in usernames:
@@ -60,6 +72,7 @@ async def search_multiple_usernames(usernames, options):
logging.error(f"Error searching username {username}: {str(e)}")
return results
def process_search_task(usernames, options, timestamp):
try:
# Setup event loop for async operations
@@ -67,7 +80,9 @@ def process_search_task(usernames, options, timestamp):
asyncio.set_event_loop(loop)
# Run the search
general_results = loop.run_until_complete(search_multiple_usernames(usernames, options))
general_results = loop.run_until_complete(
search_multiple_usernames(usernames, options)
)
# Create session folder
session_folder = os.path.join(REPORTS_FOLDER, f"search_{timestamp}")
@@ -75,7 +90,11 @@ def process_search_task(usernames, options, timestamp):
# Save the combined graph
graph_path = os.path.join(session_folder, "combined_graph.html")
maigret.report.save_graph_report(graph_path, general_results, MaigretDatabase().load_from_path(MAIGRET_DB_FILE))
maigret.report.save_graph_report(
graph_path,
general_results,
MaigretDatabase().load_from_path(MAIGRET_DB_FILE),
)
# Save individual reports
individual_reports = []
@@ -90,28 +109,49 @@ def process_search_task(usernames, options, timestamp):
context = generate_report_context(general_results)
maigret.report.save_csv_report(csv_path, username, results)
maigret.report.save_json_report(json_path, username, results, report_type='ndjson')
maigret.report.save_json_report(
json_path, username, results, report_type='ndjson'
)
maigret.report.save_pdf_report(pdf_path, context)
maigret.report.save_html_report(html_path, context)
claimed_profiles = []
for site_name, site_data in results.items():
if (site_data.get('status') and
site_data['status'].status == maigret.result.MaigretCheckStatus.CLAIMED):
claimed_profiles.append({
'site_name': site_name,
'url': site_data.get('url_user', ''),
'tags': site_data.get('status').tags if site_data.get('status') else []
})
if (
site_data.get('status')
and site_data['status'].status
== maigret.result.MaigretCheckStatus.CLAIMED
):
claimed_profiles.append(
{
'site_name': site_name,
'url': site_data.get('url_user', ''),
'tags': (
site_data.get('status').tags
if site_data.get('status')
else []
),
}
)
individual_reports.append({
'username': username,
'csv_file': os.path.join(f"search_{timestamp}", f"report_{username}.csv"),
'json_file': os.path.join(f"search_{timestamp}", f"report_{username}.json"),
'pdf_file': os.path.join(f"search_{timestamp}", f"report_{username}.pdf"),
'html_file': os.path.join(f"search_{timestamp}", f"report_{username}.html"),
'claimed_profiles': claimed_profiles,
})
individual_reports.append(
{
'username': username,
'csv_file': os.path.join(
f"search_{timestamp}", f"report_{username}.csv"
),
'json_file': os.path.join(
f"search_{timestamp}", f"report_{username}.json"
),
'pdf_file': os.path.join(
f"search_{timestamp}", f"report_{username}.pdf"
),
'html_file': os.path.join(
f"search_{timestamp}", f"report_{username}.html"
),
'claimed_profiles': claimed_profiles,
}
)
# Save results and mark job as complete
job_results[timestamp] = {
@@ -119,20 +159,19 @@ def process_search_task(usernames, options, timestamp):
'session_folder': f"search_{timestamp}",
'graph_file': os.path.join(f"search_{timestamp}", "combined_graph.html"),
'usernames': usernames,
'individual_reports': individual_reports
'individual_reports': individual_reports,
}
except Exception as e:
job_results[timestamp] = {
'status': 'failed',
'error': str(e)
}
job_results[timestamp] = {'status': 'failed', 'error': str(e)}
finally:
background_jobs[timestamp]['completed'] = True
@app.route('/')
def index():
return render_template('index.html')
@app.route('/search', methods=['POST'])
def search():
usernames_input = request.form.get('usernames', '').strip()
@@ -140,7 +179,9 @@ def search():
flash('At least one username is required', 'danger')
return redirect(url_for('index'))
usernames = [u.strip() for u in usernames_input.replace(',', ' ').split() if u.strip()]
usernames = [
u.strip() for u in usernames_input.replace(',', ' ').split() if u.strip()
]
# Create timestamp for this search session
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -157,7 +198,9 @@ def search():
# Start background job
background_jobs[timestamp] = {
'completed': False,
'thread': Thread(target=process_search_task, args=(usernames, options, timestamp))
'thread': Thread(
target=process_search_task, args=(usernames, options, timestamp)
),
}
background_jobs[timestamp]['thread'].start()
@@ -202,21 +245,23 @@ def results(session_id):
return redirect(url_for('index'))
result_data = next(
(r for r in job_results.values()
if r.get('status') == 'completed' and r['session_folder'] == session_id),
None
(
r
for r in job_results.values()
if r.get('status') == 'completed' and r['session_folder'] == session_id
),
None,
)
return render_template(
'results.html',
usernames=result_data['usernames'],
graph_file=result_data['graph_file'],
individual_reports=result_data['individual_reports'],
timestamp=session_id.replace('search_', '')
timestamp=session_id.replace('search_', ''),
)
@app.route('/reports/<path:filename>')
def download_report(filename):
try:
@@ -226,9 +271,10 @@ def download_report(filename):
logging.error(f"Error serving file {filename}: {str(e)}")
return "File not found", 404
if __name__ == '__main__':
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
app.run(debug=True)
tests/test_cli.py
+9 -4
View File
@@ -42,6 +42,7 @@ DEFAULT_ARGS: Dict[str, Any] = {
'use_disabled_sites': False,
'username': [],
'verbose': False,
'web': 5000,
'with_domains': False,
'xmind': False,
}
@@ -55,7 +56,8 @@ def test_args_search_mode(argparser):
want_args = dict(DEFAULT_ARGS)
want_args.update({'username': ['username']})
assert args == Namespace(**want_args)
for arg in vars(args):
assert getattr(args, arg) == want_args[arg]
def test_args_search_mode_several_usernames(argparser):
@@ -66,7 +68,8 @@ def test_args_search_mode_several_usernames(argparser):
want_args = dict(DEFAULT_ARGS)
want_args.update({'username': ['username1', 'username2']})
assert args == Namespace(**want_args)
for arg in vars(args):
assert getattr(args, arg) == want_args[arg]
def test_args_self_check_mode(argparser):
@@ -81,7 +84,8 @@ def test_args_self_check_mode(argparser):
}
)
assert args == Namespace(**want_args)
for arg in vars(args):
assert getattr(args, arg) == want_args[arg]
def test_args_multiple_sites(argparser):
@@ -97,4 +101,5 @@ def test_args_multiple_sites(argparser):
}
)
assert args == Namespace(**want_args)
for arg in vars(args):
assert getattr(args, arg) == want_args[arg]