function main_v15
A test function that uploads a PDF document to reMarkable cloud, syncs the local replica, and validates the upload with detailed logging and metrics.
/tf/active/vicechatdev/e-ink-llm/cloudtest/test_raw_upload.py
20 - 159
complex
Purpose
This function serves as an integration test for the reMarkable PDF upload workflow. It authenticates with reMarkable cloud, syncs the local replica to ensure current state, uploads a test PDF document, verifies the upload by checking document counts and database entries, and generates comprehensive logs and summaries. It's designed for testing and debugging the upload pipeline with full observability.
Source Code
def main():
"""Test PDF upload with detailed logging and full replica sync"""
print("š Starting PDF upload test with raw request logging...")
# Initialize logger
timestamp = int(time.time())
log_file = f"test_results/raw_upload_log_{timestamp}.log"
Path("test_results").mkdir(exist_ok=True)
print(f"š Raw requests will be logged to: {log_file}")
# Authenticate
print("š Starting reMarkable authentication...")
auth = RemarkableAuth()
session = auth.get_authenticated_session()
if not session:
print("ā Authentication failed")
return False
print("ā
Authentication complete")
# Initialize replica builder and sync current state
print("š Syncing local replica to ensure current state...")
replica_builder = RemarkableReplicaBuilder(
session=session,
replica_dir="remarkable_replica_v2"
)
# Build/sync replica
replica_builder.build_complete_replica()
# Initialize upload manager
database_path = "remarkable_replica_v2/replica_database.json"
uploader = RemarkableUploadManager(session, database_path)
# Count initial documents
with open(database_path, 'r') as f:
database = json.load(f)
initial_doc_count = sum(1 for node in database['nodes'].values()
if node['node_type'] == 'document')
print(f"š Replica location: {Path('remarkable_replica_v2').absolute()}")
# Test PDF file
test_pdf = Path("test_uploads/test_document.pdf")
if not test_pdf.exists():
print(f"ā Test PDF not found: {test_pdf}")
return False
# Upload PDF
document_name = f"RawLogTest_{timestamp}"
print(f"Uploading: {test_pdf.absolute()} as '{document_name}'")
print(f"š Initial document count: {initial_doc_count}")
success = uploader.upload_pdf_document(str(test_pdf), document_name)
if success:
print("ā
PDF upload completed successfully")
# Count final documents
with open(database_path, 'r') as f:
final_database = json.load(f)
final_doc_count = sum(1 for node in final_database['nodes'].values()
if node['node_type'] == 'document')
print(f"š Final document count: {final_doc_count}")
print(f"š Documents added: {final_doc_count - initial_doc_count}")
# Check if document exists in database
uploaded_doc = None
for uuid, node in final_database['nodes'].items():
if node.get('name') == document_name:
uploaded_doc = node
break
if uploaded_doc:
print(f"ā
Document found in database: {uploaded_doc['uuid']}")
print(f" Hash: {uploaded_doc['hash']}")
print(f" Size: {uploaded_doc.get('size', 0)} bytes")
else:
print(f"ā ļø Document not found in database with name: {document_name}")
print(f"š Check raw request log: {log_file}")
# Save detailed summary
summary = {
"timestamp": timestamp,
"document_name": document_name,
"success": True,
"log_file": log_file,
"test_pdf": str(test_pdf),
"initial_doc_count": initial_doc_count,
"final_doc_count": final_doc_count,
"documents_added": final_doc_count - initial_doc_count,
"uploaded_document": uploaded_doc
}
summary_file = f"test_results/upload_summary_{timestamp}.json"
with open(summary_file, 'w') as f:
json.dump(summary, f, indent=2)
print(f"š Summary saved to: {summary_file}")
# Test results
print("\n" + "="*50)
print("š TEST SUMMARY")
if final_doc_count > initial_doc_count:
print("PDF Upload: ā
PASS")
else:
print("PDF Upload: ā FAIL - Document count did not increase")
print(f"\nš¾ Test results saved to: {summary_file}")
else:
print("ā PDF upload failed")
# Save failure summary
summary = {
"timestamp": timestamp,
"document_name": document_name,
"success": False,
"log_file": log_file,
"test_pdf": str(test_pdf),
"initial_doc_count": initial_doc_count,
"error": "Upload failed"
}
summary_file = f"test_results/upload_summary_{timestamp}.json"
with open(summary_file, 'w') as f:
json.dump(summary, f, indent=2)
print("\n" + "="*50)
print("š TEST SUMMARY")
print("PDF Upload: ā FAIL")
print(f"\nš¾ Test results saved to: {summary_file}")
return success
Return Value
Returns a boolean value: True if the PDF upload completed successfully and the document was added to the database, False if authentication failed or the upload failed. The function also creates log files and JSON summaries in the 'test_results' directory as side effects.
Dependencies
pathlibtimejsonsys
Required Imports
import sys
from pathlib import Path
from raw_request_logger import RawRequestLogger
from auth import RemarkableAuth
from upload_manager import RemarkableUploadManager
from local_replica_v2 import RemarkableReplicaBuilder
import time
import json
Usage Example
# Ensure test PDF exists
from pathlib import Path
test_pdf_dir = Path('test_uploads')
test_pdf_dir.mkdir(exist_ok=True)
# Place a PDF file at test_uploads/test_document.pdf
# Run the test
if __name__ == '__main__':
success = main()
if success:
print('Upload test passed')
else:
print('Upload test failed')
sys.exit(0 if success else 1)
Best Practices
- Ensure the test PDF file exists at 'test_uploads/test_document.pdf' before running
- Check the generated log files in 'test_results' directory for detailed debugging information
- The function creates timestamped files to avoid conflicts between multiple test runs
- Review the upload summary JSON file for detailed metrics about the upload operation
- The function performs a full replica sync before upload, which may take time for large document collections
- Ensure sufficient disk space for the local replica in 'remarkable_replica_v2' directory
- The function modifies the local replica database, so consider backing it up before running tests
- Network connectivity is required throughout the entire operation
- Authentication credentials must be valid and not expired
Tags
Similar Components
AI-powered semantic similarity - components with related functionality:
-
function main_v6 89.3% similar
-
function main_v100 87.9% similar
-
function main_v26 85.0% similar
-
function main_v104 83.6% similar
-
function test_quick_upload_v1 83.5% similar