diff --git a/.github/workflows/publish-doc-db.yaml b/.github/workflows/publish-doc-db.yaml new file mode 100644 index 000000000..c0e08e740 --- /dev/null +++ b/.github/workflows/publish-doc-db.yaml @@ -0,0 +1,66 @@ +name: Release Documentation Database + +permissions: write-all + +on: + workflow_dispatch: # Trigger manually + +jobs: + build-and-release: + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v4 + with: + fetch-depth: 0 # Required to get all tags for versioning + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.x' + + - name: Install uv + run: | + curl -LsSf https://astral.sh/uv/install.sh | sh + echo "$HOME/.cargo/bin" >> $GITHUB_PATH + + - name: Install dependencies + run: uv pip install -r requirements.txt + + - name: Set date variable + run: echo "DATE=$(date +%F)" >> $GITHUB_ENV + + - name: Create build directory + run: mkdir -p build + + - name: Store Kotlin documentation + run: | + PYTHONPATH=scripts uv run python scripts/ingest.py -p build/documentation-db-${{ env.DATE }}.sqlite -d SourceDocs/KotlinDocs/html + PYTHONPATH=scripts uv run python scripts/ingest.py -p build/documentation-db-${{ env.DATE }}.sqlite -d SourceDocs/KotlinDocs/html/images + PYTHONPATH=scripts uv run python scripts/ingest.py -p build/documentation-db-${{ env.DATE }}.sqlite -d SourceDocs/KotlinDocs/html/frontend + PYTHONPATH=scripts uv run python scripts/ingest.py -p build/documentation-db-${{ env.DATE }}.sqlite -f SourceDocs/KotlinDocs/kotlin-spec.pdf + if [ ! -f "build/documentation-db-${{ env.DATE }}.sqlite" ]; then + echo "Failed to create database file" + exit 1 + fi + + - name: Store Java documentation + run: | + for dir in $(find SourceDocs/JavaDocs/html -type d); do + PYTHONPATH=scripts uv run python scripts/ingest.py -p build/documentation-db-${{ env.DATE }}.sqlite -d "$dir" + done + + - name: Verify database + run: | + if [ ! -f "build/documentation-db-${{ env.DATE }}.sqlite" ]; then + echo "Database file not found" + exit 1 + fi + # Add any additional verification steps here + + - name: Upload release asset + uses: softprops/action-gh-release@v1 + with: + files: build/documentation-db-${{ env.DATE }}.sqlite + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/.gitignore b/.gitignore index 8c6590038..fd796a4e8 100644 --- a/.gitignore +++ b/.gitignore @@ -6,4 +6,7 @@ ProcessDocs/ProcessAndroidDevSite/metadata.txt ProcessDocs/ProcessKotlinDocs/webhelp ProcessDocs/ProcessKotlinDocs/kotlin_writerside_docs ProcessDocs/ProcessKotlinDocs/KotlinLLMScratch/openaikey.txt -ProcessDocs/ProcessAndroidDevSite/DevsiteLLMScratch/openaikey.txt \ No newline at end of file +ProcessDocs/ProcessAndroidDevSite/DevsiteLLMScratch/openaikey.txt +__pycache__/ +*.py[cod] +*$py.class \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 9319064d6..140b18160 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,5 @@ beautifulsoup4 -lxml \ No newline at end of file +lxml +faker +brotli +Pillow diff --git a/scripts/DocumentationDatabase.py b/scripts/DocumentationDatabase.py new file mode 100644 index 000000000..2d5de30cf --- /dev/null +++ b/scripts/DocumentationDatabase.py @@ -0,0 +1,278 @@ +import os +import os.path as path +import mimetypes +import sqlite3 +import sys +import brotli +import io +from PIL import Image +import subprocess +import contextlib + +class DocumentationDatabase: + COMPRESSORS = { + 'text': 'brotli', + 'image': 'none', + 'application': 'brotli' + } + + CONTENT_TYPES = { + 'text/plain', + 'text/html', + 'text/css', + 'text/markdown', + 'image/jpeg', + 'image/png', + 'image/gif', + 'application/json', + 'application/xml', + 'font/ttf', + 'text/javascript' + } + + OVERRIDE_MIMETYPES = { + 'image/jpeg': 'none', + 'image/png': 'none', + 'image/gif': 'none', + 'image/svg+xml': 'brotli', + 'font/ttf': 'none' + } + + SCHEMA_SQL = """ + CREATE TABLE IF NOT EXISTS Content ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + path TEXT NOT NULL, + languageID INTEGER NOT NULL, + content BLOB NOT NULL, + contentTypeID INTEGER NOT NULL, + FOREIGN KEY (languageID) REFERENCES Languages(id), + FOREIGN KEY (contentTypeID) REFERENCES ContentTypes(id) + ); + + CREATE TABLE IF NOT EXISTS Languages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + value TEXT NOT NULL UNIQUE + ); + + CREATE TABLE IF NOT EXISTS ContentTypes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + value TEXT NOT NULL UNIQUE, + compression TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS ide_tooltip_table ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + path TEXT NOT NULL, + languageID INTEGER NOT NULL, + content TEXT NOT NULL, + FOREIGN KEY (languageID) REFERENCES Languages(id) + ); + """ + + def __init__(self, database_path): + self.database_path = database_path + self.input_bytes = 0 + self.stored_bytes = 0 + # Create the database if it doesn't exist or is empty + if not os.path.exists(database_path) or os.path.getsize(database_path) == 0: + with self.get_connection() as connection: + cursor = connection.cursor() + self.create_tables(cursor) + self.populate_content_types(cursor) + self.populate_languages(cursor) + connection.commit() + else: + # Check if the database conforms to the schema + with self.get_connection() as connection: + cursor = connection.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") + tables = cursor.fetchall() + expected_tables = {'ide_tooltip_table', 'Content', 'Languages', 'ContentTypes'} + existing_tables = {table[0] for table in tables} + # Ignore any tables that start with 'sqlite_' + filtered_tables = {table for table in existing_tables if not table.startswith('sqlite_')} + if filtered_tables != expected_tables: + raise ValueError("Database schema does not match the expected schema") + + @contextlib.contextmanager + def get_connection(self): + """Context manager for database connections.""" + connection = sqlite3.connect(self.database_path) + connection.execute("PRAGMA foreign_keys = ON;") # Enable foreign key constraints + try: + yield connection + finally: + connection.close() + + def get_exts(self, files): + exts = sorted({path.splitext(i)[-1] for i in files if len(path.splitext(i)[-1]) != 0}) + noexts = sorted([i for i in files if len(path.splitext(i)[-1]) == 0]) + return exts + + def create_tables(self, cursor): + cursor.executescript(self.SCHEMA_SQL) + + def populate_content_types(self, cursor): + sql = set() + for mime_type in self.CONTENT_TYPES: + major_type, minor_type = mime_type.split("/") + if mime_type in self.OVERRIDE_MIMETYPES: + compressor = self.OVERRIDE_MIMETYPES[mime_type] + elif major_type in self.COMPRESSORS: + compressor = self.COMPRESSORS[major_type] + else: + sys.exit(1) + sql.add(f"""INSERT INTO ContentTypes (value, compression) VALUES ('{mime_type}', '{compressor}');""") + # Add image/svg+xml to ContentTypes + sql.add("""INSERT INTO ContentTypes (value, compression) VALUES ('image/svg+xml', 'brotli');""") + cursor.executescript("BEGIN;\n" + "\n".join(sql) + "\nCOMMIT;\n") + + def populate_languages(self, cursor): + cursor.execute("INSERT INTO Languages (value) VALUES ('en-US');") + + def normalize_path(self, path): + """Remove leading ../ and ./ sequences from a path.""" + while path.startswith('../') or path.startswith('./'): + if path.startswith('../'): + path = path[3:] + elif path.startswith('./'): + path = path[2:] + return path + + def add_file(self, path, content, language): + with self.get_connection() as connection: + cursor = connection.cursor() + # Check if the path is a directory + if os.path.isdir(path): + print(f"Skipping directory: {path}") + return False + + # Normalize the path before processing + normalized_path = self.normalize_path(path) + + # Check if the file already exists in the database + cursor.execute("SELECT COUNT(*) FROM Content WHERE path = ?", (normalized_path,)) + if cursor.fetchone()[0] > 0: + print(f"File {normalized_path} already exists in the database. Skipping.") + return False + + # Get languageID for the given language + cursor.execute("SELECT id FROM Languages WHERE value = ?", (language,)) + language_id = cursor.fetchone()[0] + # Detect content type from file extension + ext = os.path.splitext(normalized_path)[1] + if ext not in mimetypes.types_map: + print(f"Skipping file {normalized_path}: Unsupported file extension: {ext}") + return False + content_type = mimetypes.types_map[ext] + if content_type in ['application/xml'] or ext == '.jhm': + print(f"Skipping file {normalized_path}: Unsupported content type: {content_type}") + return False + # Special handling for image files + if content_type.startswith('image/'): + if content_type == 'image/png': + # Check if the file is a valid PNG + try: + img = Image.open(io.BytesIO(content)) + if img.format != 'PNG': + print(f"Skipping file {normalized_path}: Not a valid PNG file.") + return False + except Exception as e: + print(f"Skipping file {normalized_path}: Error checking PNG format: {e}") + return False + # Call pngquant in a subshell + process = subprocess.Popen(['pngquant', '--force', '--output', '-', '-'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + stdout, stderr = process.communicate(input=content) + if process.returncode != 0: + raise RuntimeError(f"pngquant failed: {stderr.decode()}") + compressed_content = stdout + elif content_type in self.OVERRIDE_MIMETYPES: + # Use the compression method specified in OVERRIDE_MIMETYPES + compressor = self.OVERRIDE_MIMETYPES[content_type] + if compressor == 'brotli': + compressed_content = brotli.compress(content) + else: + compressed_content = content + else: + compressed_content = content + else: + # Compress non-image files + compressed_content = brotli.compress(content) + # Get contentTypeID for the detected content type + cursor.execute("SELECT id FROM ContentTypes WHERE value = ?", (content_type,)) + content_type_id = cursor.fetchone() + if content_type_id is None: + print(f"Content type {content_type} not found in ContentTypes table.") + return False + content_type_id = content_type_id[0] + # Insert the file into the Content table + cursor.execute( + "INSERT INTO Content (path, languageID, content, contentTypeID) VALUES (?, ?, ?, ?)", + (normalized_path, language_id, compressed_content, content_type_id) + ) + # Update byte counters + self.input_bytes += len(content) + self.stored_bytes += len(compressed_content) + connection.commit() + return True + + def get_file(self, path, language): + with self.get_connection() as connection: + cursor = connection.cursor() + # Get languageID for the given language + cursor.execute("SELECT id FROM Languages WHERE value = ?", (language,)) + language_id = cursor.fetchone()[0] + # Retrieve the file content and content type from the Content table + cursor.execute("SELECT content, contentTypeID FROM Content WHERE path = ? AND languageID = ?", (path, language_id)) + result = cursor.fetchone() + if result is None: + raise FileNotFoundError(f"File not found: {path} for language: {language}") + content, content_type_id = result + # Get the content type + cursor.execute("SELECT value FROM ContentTypes WHERE id = ?", (content_type_id,)) + content_type = cursor.fetchone()[0] + # Decompress the content if necessary + if content_type.startswith('image/'): + # Image files are not compressed + return io.BytesIO(content) + else: + # Decompress non-image files + return io.BytesIO(brotli.decompress(content)) + + def emit_summary(self, label=None): + """ + Prints a summary with the total count of files stored and the number of files grouped by each content type. + If a label is provided, it will be printed at the start of the method. + """ + if label: + print(label) + with self.get_connection() as connection: + cursor = connection.cursor() + # Total count of files + cursor.execute("SELECT COUNT(*) FROM Content") + total_files = cursor.fetchone()[0] + print(f"Total files stored: {total_files}") + + # Number of files grouped by content type + cursor.execute(''' + SELECT ContentTypes.value, COUNT(*) + FROM Content + JOIN ContentTypes ON Content.contentTypeID = ContentTypes.id + GROUP BY ContentTypes.value + ''') + print("Files by content type:") + for mime_type, count in cursor.fetchall(): + print(f" {mime_type}: {count}") + + def stats(self): + with self.get_connection() as connection: + cursor = connection.cursor() + # Get count of files + cursor.execute("SELECT COUNT(*) FROM Content") + count = cursor.fetchone()[0] + return count, self.input_bytes, self.stored_bytes + + # Removed write_languages() method + +mimetypes.types_map[".svg"] = "image/svg+xml" +mimetypes.types_map[".ttf"] = "font/ttf" \ No newline at end of file diff --git a/scripts/create_empty_database.py b/scripts/create_empty_database.py new file mode 100644 index 000000000..89ddec387 --- /dev/null +++ b/scripts/create_empty_database.py @@ -0,0 +1,12 @@ +#! /usr/bin/python + +import sys +from DocumentationDatabase import DocumentationDatabase + +if __name__ == "__main__": + if len(sys.argv) != 2: + print("Usage: python create_empty_database.py ") + sys.exit(1) + database_path = sys.argv[1] + db = DocumentationDatabase(database_path) + db.create_empty_database("/tmp/files") diff --git a/scripts/gen_fake.py b/scripts/gen_fake.py new file mode 100644 index 000000000..41fd72938 --- /dev/null +++ b/scripts/gen_fake.py @@ -0,0 +1,65 @@ +from faker import Faker +import os +import random + +fake = Faker() +output_dir = "fake_files" + +if not os.path.exists(output_dir): + os.makedirs(output_dir) + +file_types = ["txt", "md", "html", "css", "png", "jpg"] + +for i in range(100): + file_type = random.choice(file_types) + filename = f"fake_file_{i+1}.{file_type}" + filepath = os.path.join(output_dir, filename) + + if file_type == "txt": + content = fake.paragraph(nb_sentences=random.randint(5, 15)) + with open(filepath, "w") as f: + f.write(content) + print(f"Generated: {filename} (text)") + elif file_type == "md": + title = fake.sentence() + paragraphs = [fake.paragraph(nb_sentences=random.randint(3, 8)) for _ in range(random.randint(2, 5))] + content = f"# {title}\n\n" + "\n\n".join(paragraphs) + with open(filepath, "w") as f: + f.write(content) + print(f"Generated: {filename} (markdown)") + elif file_type == "html": + title = fake.sentence() + body = "\n ".join([f"

{fake.paragraph(nb_sentences=random.randint(2, 5))}

" for _ in range(random.randint(3, 7))]) + content = f""" + + + + {title} + + + {body} + +""" + with open(filepath, "w") as f: + f.write(content) + print(f"Generated: {filename} (html)") + elif file_type == "css": + selectors = [fake.word() for _ in range(random.randint(2, 5))] + rules = [] + content = "" # Initialize content variable + for selector in selectors: + num_rules = random.randint(1, 3) + for _ in range(num_rules): + property_name = random.choice(["color", "font-size", "margin", "padding", "background-color"]) + property_value = fake.word() if property_name in ["color"] else f"{random.randint(10, 20)}px" + rules.append(f" {property_name}: {property_value};") + content += f"{selector} {{\n" + "\n".join(rules) + "\n}\n\n" + with open(filepath, "w") as f: + f.write(content) + print(f"Generated: {filename} (css)") + elif file_type == "png" or file_type == "jpg": + # For image files, we'll just create an empty file for simplicity + open(filepath, 'a').close() + print(f"Generated: {filename} (empty image file)") + +print(f"\nSuccessfully generated 100 fake files in the '{output_dir}' directory.") diff --git a/scripts/ingest.py b/scripts/ingest.py new file mode 100644 index 000000000..8371d2711 --- /dev/null +++ b/scripts/ingest.py @@ -0,0 +1,38 @@ +import argparse +import os +import multiprocessing +from DocumentationDatabase import DocumentationDatabase + +def process_file(file_path, db_path): + print(f"Processing file: {file_path}") + + # Add file to database + db = DocumentationDatabase(db_path) + with open(file_path, 'rb') as file: + content = file.read() + if db.add_file(file_path, content, 'en-US'): + print(f"Added file {file_path} to the database.") + +def main(): + parser = argparse.ArgumentParser(description='Ingest files into a documentation database.') + parser.add_argument('-f', '--file', help='Path to a file to add to the database.') + parser.add_argument('-d', '--directory', help='Path to a directory to add all files to the database.') + parser.add_argument('-p', '--path-to-db', help='Path to the SQLite database file.', default='/tmp/my-doc-db.sqlite') + args = parser.parse_args() + + db_path = args.path_to_db + db = DocumentationDatabase(db_path) + + if args.file: + with open(args.file, 'rb') as file: + content = file.read() + if db.add_file(args.file, content, 'en-US'): + print(f"Added file {args.file} to the database.") + + if args.directory: + file_paths = [os.path.join(args.directory, filename) for filename in os.listdir(args.directory) if os.path.isfile(os.path.join(args.directory, filename))] + with multiprocessing.Pool() as pool: + pool.starmap(process_file, [(file_path, db_path) for file_path in file_paths]) + +if __name__ == '__main__': + main() diff --git a/scripts/list_database_documents.py b/scripts/list_database_documents.py new file mode 100644 index 000000000..6ea1e4a8d --- /dev/null +++ b/scripts/list_database_documents.py @@ -0,0 +1,29 @@ +import argparse +import sqlite3 +import os + +def main(): + parser = argparse.ArgumentParser(description='List documents in a SQLite database.') + parser.add_argument('-p', '--path', help='Path to the SQLite database file.', required=True) + args = parser.parse_args() + db_path = args.path + print(f"Database path: {db_path}") + + if not os.path.exists(db_path): + print(f"Error: Database file not found at {db_path}") + return + + try: + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + cursor.execute("SELECT path, LENGTH(content) FROM content") + for row in cursor.fetchall(): + print(f"Path: {row[0]}, Bytes: {row[1]}") + except sqlite3.Error as e: + print(f"Database error: {e}") + finally: + if 'conn' in locals(): + conn.close() + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/scripts/main.py b/scripts/main.py new file mode 100644 index 000000000..57bcadc20 --- /dev/null +++ b/scripts/main.py @@ -0,0 +1,6 @@ +def main(): + print("Hello from doc-db!") + + +if __name__ == "__main__": + main() diff --git a/scripts/myServer.py b/scripts/myServer.py new file mode 100644 index 000000000..da27a09d0 --- /dev/null +++ b/scripts/myServer.py @@ -0,0 +1,160 @@ +#! /usr/bin/python + +import os.path as path +import mimetypes +import http +from http.server import BaseHTTPRequestHandler, HTTPServer +import sqlite3 +import sys +import time +import threading +import argparse + +DATABASE_FILENAME = "Documentation.db" +HOST_NAME = "localhost" +SERVER_PORT = 8080 #6174 +QUERY_SQL = """SELECT + content, + CT.value as contentType, + CT.compression as compression +FROM + Content as C +INNER JOIN Languages as L ON C.languageID = L.ID +INNER JOIN ContentTypes as CT ON C.contentTypeID = CT.ID +WHERE + C.path = ? +AND L.value = ?;""" + +DEFAULT_LANGUAGE = "en-US" +ACCEPT_LANGUAGES = "Accept-Language" # Client's acceptable languages +ACCEPT_ENCODINGS = "Accept-Encoding" # Client's acceptable compression types +ACCEPT = "Accept" # Client's acceptable content types +BROTLI_TAG = "br" +BROTLI_COMPRESSION = "brotli" + +# Thread-local storage for database connections +thread_local = threading.local() + +def get_db(): + if not hasattr(thread_local, "connection"): + thread_local.connection = sqlite3.connect(DATABASE_FILENAME) + thread_local.cursor = thread_local.connection.cursor() + return thread_local.cursor + +# ------------------------------------------------------------------------------ +def get_header(headers, key, default_value): + return [item.strip() for item in headers.get(key, default_value).split(",")] + + + +# ------------------------------------------------------------------------------ +def get_client_languages(headers): + # TODO: Get the language with the highest q factor. --DS, 6-May-2025 + + return get_header(headers, ACCEPT_LANGUAGES, DEFAULT_LANGUAGE) + + + +# ------------------------------------------------------------------------------ +def format_error(error, url_path, language): + message = b"Error %s %s%s
%s
%s" % \ + (bytes(str(error.value), "utf-8"), + bytes(error.phrase, "utf-8"), + bytes(f"""Path: {url_path}""", "utf-8"), + bytes(f"""Language: {language}""", "utf-8"), + bytes(f"""""", "utf-8") + ) + + return error, (message, 'text/html', 'None') + + + +# ------------------------------------------------------------------------------ +def get_data(cursor, url_path, language): + cursor.execute(QUERY_SQL, (url_path, language)) + rows = cursor.fetchall() + + print(f"""In get_data(), url_path='{url_path}', language='{language}' and there are {len(rows)} rows.""") + + if len(rows) == 1: + result = http.HTTPStatus.OK, rows[0] + + elif len(rows) == 0: + print(f"""DATABASE CONTENT ERROR: there is no match for url_path='{url_path}'.""") + result = format_error(http.HTTPStatus.NOT_FOUND, url_path, language) + + else: # len(rows) > 1 + print(f"""DATABASE CONTENT ERROR: for url_path='{url_path}', there are {len(rows)}, not 1.""") + result = format_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, url_path, language) + + return result + + + +# ------------------------------------------------------------------------------ +def client_supports_brotli(headers): + return BROTLI_TAG in get_header(headers, ACCEPT_ENCODINGS, "") + + + +# ------------------------------------------------------------------------------ +def uncompress_brotli(content): + print("TODO: Replace this function with Brotli decompression. --DS, 6-May-2025") + + return b"""Brotli decompression is required but the server doesn't do it yet. Sorry.\n""" + + + +# ------------------------------------------------------------------------------ +class MyServer(BaseHTTPRequestHandler): + def do_GET(self): + cursor = get_db() + status, (content, content_type, compression) = \ + get_data(cursor, self.path[1:], get_client_languages(self.headers)[0]) + + # TODO: Test that the client will accept the content type we propose. Use ACCEPT for that. --DS, 6-May-2025 + + from pprint import pformat; print(f"""Headers: {pformat(self.headers.items())}""") + if compression == BROTLI_COMPRESSION and client_supports_brotli(self.headers) == False: + content = uncompress_brotli(content) + compression = "None" + + self.send_response(status.value) + self.send_header("Content-type", content_type) + self.send_header("Content-length", len(content)) + if compression == BROTLI_COMPRESSION: + self.send_header("Content-encoding", BROTLI_TAG) + self.end_headers() + self.wfile.write(content) + + + +# ------------------------------------------------------------------------------ +def parse_arguments(): + parser = argparse.ArgumentParser(description='Start the documentation server.') + parser.add_argument('-d', '--database', type=str, default="Documentation.db", + help='Path to the SQLite database file (default: Documentation.db)') + parser.add_argument('-p', '--port', type=int, default=8080, + help='Port to run the server on (default: 8080)') + return parser.parse_args() + +def start_server(database_filename, server_port): + myServer = HTTPServer((HOST_NAME, server_port), MyServer) + print(f"Server started http://{HOST_NAME}:{server_port}") + + try: + myServer.serve_forever() + except KeyboardInterrupt: + pass + + myServer.server_close() + print("Server stopped.") + + + +# ------------------------------------------------------------------------------ +# ------------------------------------------------------------------------------ +if __name__ == "__main__": + args = parse_arguments() + DATABASE_FILENAME = args.database + start_server(DATABASE_FILENAME, args.port) diff --git a/scripts/put_content_in_database.py b/scripts/put_content_in_database.py new file mode 100644 index 000000000..c234221c8 --- /dev/null +++ b/scripts/put_content_in_database.py @@ -0,0 +1,196 @@ +#! /usr/bin/python + +import os +import os.path as path +import mimetypes +import sqlite3 +import subprocess +import sys + + +LIST_OF_FILES = "/tmp/files" +LIST_OF_DIRECTORIES = "/tmp/directories" +DATABASE_FILENAME = "/home/david/Documentation.db" +CLONE_INTO_DIR = "../cloneContentDir" +COMPRESSION_PARALLELISM = os.cpu_count() - 1 if os.cpu_count() else 1 + +SKIP_EXTS = {"", + ".version", + } + +OVERRIDE_EXTS = {"tar" : "gzip", + } + +OVERRIDE_MIMETYPES = {"image/svg+xml" : "brotli", + } + +mimetypes.types_map[".md"] = "text/plain" +mimetypes.types_map[".log"] = "text/plain" +mimetypes.types_map[".dtd"] = "text/plain" +mimetypes.types_map[".Debian"] = "text/plain" +mimetypes.types_map[".alternatives"] = "text/plain" + +INSERT_SQL = """INSERT INTO Content (path, languageID, content, contentTypeID) VALUES (?, 1, '', ?);""" +UPDATE_SQL = """UPDATE Content SET content=? WHERE path=? AND languageID=1;""" + + +# ------------------------------------------------------------------------------ +def clone_directories(clone_into_dir): + with open(LIST_OF_DIRECTORIES) as fd: + dirnames = [i.strip()[2:] for i in fd][1:] # Skip the first, top directory. + + os.mkdir(clone_into_dir) + [os.mkdir(path.join(clone_into_dir, d)) for d in dirnames] + + + +# ------------------------------------------------------------------------------ +def get_tables(cursor): + rows = cursor.execute("""SELECT value, id, compression FROM ContentTypes ORDER BY id;""").fetchall() + + content_type_table = {row[0] : int(row[1]) for row in rows} + compression_table = {row[0] : row[2] for row in rows} + + return content_type_table, compression_table + + + +# ------------------------------------------------------------------------------ +def get_compression_command(input_filename, output_filename, compressor): + if compressor == "brotli": + compression_command = f"""brotli --best -o '{output_filename}' '{input_filename}'""" + + elif compressor == "None": + compression_command = f"""cp '{input_filename}' '{output_filename}'""" + + else: + print(f"""Bad compressor, '{compressor}' for file '{input_filename}'.\n\n""") + sys.exit(1) + + return compression_command + + + +# ------------------------------------------------------------------------------ +def compress_noexts(compression_table, clone_into_dir, pathnames): + noexts = [pathname for pathname in pathnames if len(path.splitext(pathname)[-1]) == 0] + + compressor = compression_table["text/plain"] + + return [get_compression_command(pathname, path.join(clone_into_dir, pathname), compressor) + for pathname in noexts] + + + +# ------------------------------------------------------------------------------ +def compress_exts(compression_table, clone_into_dir, pathnames): + compression_commands = [ ] + + for input_pathname in pathnames: + if path.splitext(input_pathname)[-1] in SKIP_EXTS: + continue + + mimetype = mimetypes.types_map[path.splitext(input_pathname)[-1]] + output_pathname = path.join(clone_into_dir, input_pathname) + + compression_commands.append(get_compression_command(input_pathname, output_pathname, compression_table[mimetype])) + + return compression_commands + + + +# ------------------------------------------------------------------------------ +def compress(commands): + number_of_commands = len(commands) + jobs = [] + + interval = 1 + int(1.0 * len(commands) / COMPRESSION_PARALLELISM) + + for i in range(COMPRESSION_PARALLELISM): + items = commands[0:interval] + commands = commands[interval:] + + jobs.append(subprocess.Popen(args=";".join(items), shell=True)) + + print(f"There are {number_of_commands} compression commands and {len(jobs)} parallel jobs. This will take a few minutes.") + + [job.wait() for job in jobs] + + + +# ------------------------------------------------------------------------------ +def insert_noexts(content_type_table, clone_into_dir, pathnames): + noexts = [pathname for pathname in pathnames if len(path.splitext(pathname)[-1]) == 0] + + content_type_id = content_type_table["text/plain"] + + return [(pathname, content_type_id, path.join(clone_into_dir, pathname)) + for pathname in noexts] + + + +# ------------------------------------------------------------------------------ +def insert_exts(content_type_table, clone_into_dir, pathnames): + inserts = [] + + for input_pathname in pathnames: + if path.splitext(input_pathname)[-1] in SKIP_EXTS: + continue + + mimetype = mimetypes.types_map[path.splitext(input_pathname)[-1]] + output_pathname = path.join(clone_into_dir, input_pathname) + + inserts.append((input_pathname, content_type_table[mimetype], output_pathname)) + + return inserts + + + +# ------------------------------------------------------------------------------ +def write_to_database(cursor, sql): + for pathname, content_type_id, compressed_path in sql: + cursor.execute(INSERT_SQL, (pathname, content_type_id)) + + with open(compressed_path, "rb") as fd: + data = fd.read() + + cursor.execute(UPDATE_SQL, (sqlite3.Binary(data), pathname)) + + cursor.execute(INSERT_SQL, ('x.html', 3)) + cursor.execute(UPDATE_SQL, (sqlite3.Binary(b'Hello, Jim'), 'x.html')) + + + +# ------------------------------------------------------------------------------ +def main(cursor): + with open(LIST_OF_FILES) as fd: + pathnames = [i.strip()[2:] for i in fd] + + clone_directories(CLONE_INTO_DIR) + print(f"Directories cloned into '{CLONE_INTO_DIR}'.") + + content_type_table, compression_table = get_tables(cursor) + + compression_commands = [] + compression_commands.extend(compress_exts( compression_table, CLONE_INTO_DIR, pathnames)) + compression_commands.extend(compress_noexts(compression_table, CLONE_INTO_DIR, pathnames)) + + compress(compression_commands) + + sql = [] + sql.extend(insert_exts( content_type_table, CLONE_INTO_DIR, pathnames)) + sql.extend(insert_noexts(content_type_table, CLONE_INTO_DIR, pathnames)) + + cursor.execute("PRAGMA foreign_keys = ON;") # Enable referential integrity enforcement; + cursor.execute("DELETE FROM Content;") # Delete old content. + + print("Starting database inserts.") + write_to_database(cursor, sql) + + + +# ------------------------------------------------------------------------------ +# ------------------------------------------------------------------------------ +if __name__ == "__main__": + with sqlite3.connect(DATABASE_FILENAME) as connection: + main(connection.cursor()) diff --git a/scripts/put_json_in_database.py b/scripts/put_json_in_database.py new file mode 100644 index 000000000..95a62d3b8 --- /dev/null +++ b/scripts/put_json_in_database.py @@ -0,0 +1,114 @@ +#! /usr/bin/python + +import json +import os +import os.path as path +from pprint import pformat +import sqlite3 +import sys + + +JSON_FILE = "CoGoTooltips.json" +DATABASE_FILENAME = "/home/david/Documentation.db" +TOOLTIP_TABLE = "ide_tooltip_table" +TOOLTIP_COLUMNS = ("tooltipTag", "tooltipCategory", "tooltipSummary", "tooltipDetail", "tooltipButtons") +INSERT_SQL = f"""INSERT INTO {TOOLTIP_TABLE} ({", ".join(TOOLTIP_COLUMNS)}) VALUES (?, ?, ?, ?, ?);""" + + + +# ------------------------------------------------------------------------------ +# A sample item: +# +# {'buttons': [['Learn more about module java.base', +# 'file:///android_asset/CoGoTooltips/external/javadocs/api/java.base/module-summary.html']], +# 'category': 'java', +# 'detail': 'Defines the foundational APIs of the Java SE Platform.\n' +# '\n' +# '
Providers:
The JDK implementation ' +# 'of this module provides an implementation of the jrt file system provider to enumerate and read ' +# 'the class and resource files in a run-time image. The jrt file ' +# 'system can be created by calling FileSystems.newFileSystem(URI.create("jrt:/")). ' +# '
', +# 'summary': 'Defines the foundational APIs of the Java SE Platform.', +# 'tag': 'java.base'} + +def get_insert(item): + buttons = [{"first" : button[0], + "second" : button[1]} for button in item["buttons"]] + + return {"tag" : item["tag"], + "category": item["category"], + "summary" : item["summary"], + "detail" : item["detail"], + "buttons" : json.dumps(buttons, indent=None)} + + + +# ------------------------------------------------------------------------------ +def get_inserts(json_items): + uniques = {} + for python_item in [get_insert(json_item) for json_item in json_items]: + + primary_key = python_item["tag"], python_item["category"] + + if primary_key in uniques: + print(f"""WARNING: Discarding duplicate '{primary_key}' item.""") + + else: + uniques[primary_key] = python_item + + print(f"Eliminated {len(json_items) - len(uniques)} duplicates.") + + return tuple(uniques.values()) + + + +# ------------------------------------------------------------------------------ +def report_bad_items(items, keyword, failure_mode): + print(f""" +{"-" * 80} +WARNING: {len(items)} items were eliminated because the '{keyword}' element is {failure_mode}. +Items missing the '{keyword}' element are:\n""") + print(f"""{"\n\n".join([pformat(after_item) for after_item in items])}""") + + + +# ------------------------------------------------------------------------------ +def main(cursor): + with open(JSON_FILE) as fd: + items = json.load(fd) + + print(f"Read {len(items)} JSON entries from '{JSON_FILE}'.") + + inserts = get_inserts(items) + + inserts2 = [insert for insert in inserts if type(insert["detail"]) == type("")] + bad_items = [insert for insert in inserts if type(insert["detail"]) != type("")] + report_bad_items(bad_items, "detail", "not a string") + + inserts3 = [insert for insert in inserts2 if len(insert["detail"]) > 0] + bad_items = [insert for insert in inserts if len(insert["detail"]) <= 0] + report_bad_items(bad_items, "detail", "an empty string") + + inserts4 = [insert for insert in inserts3 if len(insert["summary"]) > 0] + bad_items = [insert for insert in inserts if len(insert["summary"]) <= 0] + report_bad_items(bad_items, "summary", "an empty string") + + inserts = [(insert["tag"], insert["category"], insert["summary"], insert["detail"], insert["buttons"]) + for insert in inserts4] + + cursor.execute("PRAGMA foreign_keys = ON;") # Enable referential integrity enforcement. + cursor.execute("DELETE FROM ide_tooltip_table;") # Delete old content. + cursor.executemany(INSERT_SQL, inserts) + + + +# ------------------------------------------------------------------------------ +# ------------------------------------------------------------------------------ +if __name__ == "__main__": + with sqlite3.connect(DATABASE_FILENAME) as connection: + main(connection.cursor()) diff --git a/scripts/test_add_file.py b/scripts/test_add_file.py new file mode 100644 index 000000000..1160060ee --- /dev/null +++ b/scripts/test_add_file.py @@ -0,0 +1,381 @@ +import unittest +import os +import tempfile +import sqlite3 +from DocumentationDatabase import DocumentationDatabase +from faker import Faker +from PIL import Image +import io +from faker_file.providers.jpeg_file import JpegFileProvider +from faker_file.providers.image.pil_generator import PilImageGenerator +from faker_file.storages.filesystem import FileSystemStorage + +class TestAddFile(unittest.TestCase): + def setUp(self): + # Create a temporary database file name for testing, but remove the file so DocumentationDatabase can create it + self.temp_db_file = tempfile.NamedTemporaryFile(delete=False) + self.temp_db_file.close() + os.unlink(self.temp_db_file.name) + self.db = DocumentationDatabase(self.temp_db_file.name) + + def tearDown(self): + # Clean up the temporary database file if it exists + if os.path.exists(self.temp_db_file.name): + os.unlink(self.temp_db_file.name) + + def test_add_html_file_to_database(self): + # Insert an HTML file into the Content table using a missing method + html_path = 'testfile.html' + html_content = b'Hello' + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Count the number of files before adding the new file + cursor.execute("SELECT COUNT(*) FROM Content") + count_before = cursor.fetchone()[0] + # Call the missing method to add the file + self.db.add_file(html_path, html_content, 'en-US') + # Verify the file is present and the number of files increased by 1 + cursor.execute("SELECT COUNT(*) FROM Content") + count_after = cursor.fetchone()[0] + self.assertEqual(count_after, count_before + 1) + + def test_add_random_html_file_to_database(self): + fake = Faker() + # Generate a random HTML file of at least 5 KB + html_content = b'' + fake.sentence().encode() + b'' + while len(html_content) < 5 * 1024: # Ensure at least 5 KB + html_content += b'

' + fake.paragraph().encode() + b'

' + html_content += b'' + html_path = 'random_' + fake.file_name(extension='html') + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Count the number of files before adding the new file + cursor.execute("SELECT COUNT(*) FROM Content") + count_before = cursor.fetchone()[0] + # Call the missing method to add the file + self.db.add_file(html_path, html_content, 'en-US') + # Verify the file is present and the number of files increased by 1 + cursor.execute("SELECT COUNT(*) FROM Content") + count_after = cursor.fetchone()[0] + self.assertEqual(count_after, count_before + 1) + + def test_compressed_content_is_smaller(self): + fake = Faker() + # Generate a large HTML file to ensure compression is noticeable + html_content = b'' + fake.sentence().encode() + b'' + while len(html_content) < 10 * 1024: # Ensure at least 10 KB + html_content += b'

' + fake.paragraph().encode() + b'

' + html_content += b'' + html_path = 'compressed_' + fake.file_name(extension='html') + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Call the missing method to add the file + self.db.add_file(html_path, html_content, 'en-US') + # Verify the stored content is smaller than the original + cursor.execute("SELECT content FROM Content WHERE path = ?", (html_path,)) + stored_content = cursor.fetchone()[0] + self.assertLess(len(stored_content), len(html_content)) + + def test_add_plain_text_file_to_database(self): + fake = Faker() + # Generate a large plain text file to ensure compression is noticeable + text_content = b'' + while len(text_content) < 10 * 1024: # Ensure at least 10 KB + text_content += fake.paragraph().encode() + b'\n' + text_path = 'random_' + fake.file_name(extension='txt') + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Count the number of files before adding the new file + cursor.execute("SELECT COUNT(*) FROM Content") + count_before = cursor.fetchone()[0] + # Call the missing method to add the file + self.db.add_file(text_path, text_content, 'en-US') + # Verify the file is present and the number of files increased by 1 + cursor.execute("SELECT COUNT(*) FROM Content") + count_after = cursor.fetchone()[0] + self.assertEqual(count_after, count_before + 1) + # Verify the stored content is smaller than the original + cursor.execute("SELECT content FROM Content WHERE path = ?", (text_path,)) + stored_content = cursor.fetchone()[0] + self.assertLess(len(stored_content), len(text_content)) + + def test_add_css_file_to_database(self): + fake = Faker() + # Generate a large CSS file to ensure compression is noticeable + css_content = b'' + while len(css_content) < 10 * 1024: # Ensure at least 10 KB + css_content += f""" + .{fake.word()} {{ + color: {fake.hex_color()}; + font-size: {fake.random_int(min=12, max=24)}px; + margin: {fake.random_int(min=0, max=20)}px; + padding: {fake.random_int(min=0, max=20)}px; + background-color: {fake.hex_color()}; + }} + """.encode() + css_path = 'random_' + fake.file_name(extension='css') + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Count the number of files before adding the new file + cursor.execute("SELECT COUNT(*) FROM Content") + count_before = cursor.fetchone()[0] + # Call the missing method to add the file + self.db.add_file(css_path, css_content, 'en-US') + # Verify the file is present and the number of files increased by 1 + cursor.execute("SELECT COUNT(*) FROM Content") + count_after = cursor.fetchone()[0] + self.assertEqual(count_after, count_before + 1) + # Verify the stored content is smaller than the original + cursor.execute("SELECT content FROM Content WHERE path = ?", (css_path,)) + stored_content = cursor.fetchone()[0] + self.assertLess(len(stored_content), len(css_content)) + + def test_add_markdown_file_to_database(self): + fake = Faker() + # Generate a large Markdown file to ensure compression is noticeable + markdown_content = b'' + while len(markdown_content) < 10 * 1024: # Ensure at least 10 KB + markdown_content += f""" + # {fake.sentence()} + {fake.paragraph()} + ## {fake.sentence()} + {fake.paragraph()} + """.encode() + markdown_path = 'random_' + fake.file_name(extension='md') + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Count the number of files before adding the new file + cursor.execute("SELECT COUNT(*) FROM Content") + count_before = cursor.fetchone()[0] + # Call the missing method to add the file + self.db.add_file(markdown_path, markdown_content, 'en-US') + # Verify the file is present and the number of files increased by 1 + cursor.execute("SELECT COUNT(*) FROM Content") + count_after = cursor.fetchone()[0] + self.assertEqual(count_after, count_before + 1) + # Verify the stored content is smaller than the original + cursor.execute("SELECT content FROM Content WHERE path = ?", (markdown_path,)) + stored_content = cursor.fetchone()[0] + self.assertLess(len(stored_content), len(markdown_content)) + + def test_add_jpeg_image_to_database(self): + fake = Faker() + # Create a simple JPEG image using Pillow + img = Image.new('RGB', (800, 600), color='red') + img_byte_arr = io.BytesIO() + img.save(img_byte_arr, format='JPEG') + jpeg_content = img_byte_arr.getvalue() + jpeg_path = 'random_' + fake.file_name(extension='jpg') + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Count the number of files before adding the new file + cursor.execute("SELECT COUNT(*) FROM Content") + count_before = cursor.fetchone()[0] + # Call the missing method to add the file + self.db.add_file(jpeg_path, jpeg_content, 'en-US') + # Verify the file is present and the number of files increased by 1 + cursor.execute("SELECT COUNT(*) FROM Content") + count_after = cursor.fetchone()[0] + self.assertEqual(count_after, count_before + 1) + # Verify the stored content is not compressed + cursor.execute("SELECT content FROM Content WHERE path = ?", (jpeg_path,)) + stored_content = cursor.fetchone()[0] + self.assertEqual(len(stored_content), len(jpeg_content)) + + def test_add_gif_image_to_database(self): + fake = Faker() + # Create a simple GIF image using Pillow + img = Image.new('RGB', (800, 600), color='blue') + img_byte_arr = io.BytesIO() + img.save(img_byte_arr, format='GIF') + gif_content = img_byte_arr.getvalue() + gif_path = 'random_' + fake.file_name(extension='gif') + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Count the number of files before adding the new file + cursor.execute("SELECT COUNT(*) FROM Content") + count_before = cursor.fetchone()[0] + # Call the missing method to add the file + self.db.add_file(gif_path, gif_content, 'en-US') + # Verify the file is present and the number of files increased by 1 + cursor.execute("SELECT COUNT(*) FROM Content") + count_after = cursor.fetchone()[0] + self.assertEqual(count_after, count_before + 1) + # Verify the stored content is not compressed + cursor.execute("SELECT content FROM Content WHERE path = ?", (gif_path,)) + stored_content = cursor.fetchone()[0] + self.assertEqual(len(stored_content), len(gif_content)) + + def test_add_png_image_to_database(self): + fake = Faker() + # Create a PNG image: green on the left half, yellow on the right half + width, height = 800, 600 + img = Image.new('RGB', (width, height)) + for x in range(width): + for y in range(height): + if x < width // 2: + img.putpixel((x, y), (0, 128, 0)) # green + else: + img.putpixel((x, y), (255, 255, 0)) # yellow + img_byte_arr = io.BytesIO() + img.save(img_byte_arr, format='PNG') + png_content = img_byte_arr.getvalue() + png_path = 'random_' + fake.file_name(extension='png') + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Count the number of files before adding the new file + cursor.execute("SELECT COUNT(*) FROM Content") + count_before = cursor.fetchone()[0] + # Call the missing method to add the file + self.db.add_file(png_path, png_content, 'en-US') + # Verify the file is present and the number of files increased by 1 + cursor.execute("SELECT COUNT(*) FROM Content") + count_after = cursor.fetchone()[0] + self.assertEqual(count_after, count_before + 1) + # Verify the stored content is less than or equal to the original + cursor.execute("SELECT content FROM Content WHERE path = ?", (png_path,)) + stored_content = cursor.fetchone()[0] + self.assertLessEqual(len(stored_content), len(png_content)) + + def test_add_svg_image_to_database(self): + # Create a more complex SVG content + svg_content = b''' + + + + + + + + + + + SVG Test + + + + ''' + svg_path = 'test_svg.svg' + + # Add the SVG file to the database + self.db.add_file(svg_path, svg_content, 'en-US') + + # Verify the file was added correctly + with sqlite3.connect(self.db.database_path) as connection: + cursor = connection.cursor() + cursor.execute("SELECT content FROM Content WHERE path = ?", (svg_path,)) + result = cursor.fetchone() + self.assertIsNotNone(result) + # The stored content should be brotli-compressed, so it should be smaller than the original + self.assertLess(len(result[0]), len(svg_content)) + + def test_add_version_file_to_database(self): + fake = Faker() + # Generate a random version file + version_content = fake.text().encode() + version_path = 'random_' + fake.file_name(extension='version') + + # Create a temporary database file + temp_db_file = tempfile.NamedTemporaryFile(delete=False) + temp_db_file.close() + os.unlink(temp_db_file.name) + + # Initialize the database + db = DocumentationDatabase(temp_db_file.name) + + # Get initial file count and byte totals + with sqlite3.connect(temp_db_file.name) as connection: + cursor = connection.cursor() + cursor.execute("SELECT COUNT(*) FROM Content") + initial_count = cursor.fetchone()[0] + initial_input_bytes = db.input_bytes + initial_stored_bytes = db.stored_bytes + + # Attempt to add the version file to the database + db.add_file(version_path, version_content, 'en-US') + + # Verify the file count and byte totals remain unchanged + with sqlite3.connect(temp_db_file.name) as connection: + cursor = connection.cursor() + cursor.execute("SELECT COUNT(*) FROM Content") + final_count = cursor.fetchone()[0] + self.assertEqual(final_count, initial_count) + self.assertEqual(db.input_bytes, initial_input_bytes) + self.assertEqual(db.stored_bytes, initial_stored_bytes) + + # Clean up the temporary database file + os.unlink(temp_db_file.name) + + def test_skip_jpeg_as_png_to_database(self): + fake = Faker() + # Create a JPEG image + img = Image.new('RGB', (800, 600), color='red') + img_byte_arr = io.BytesIO() + img.save(img_byte_arr, format='JPEG') + jpeg_content = img_byte_arr.getvalue() + # Save it with a .png extension + png_path = 'random_' + fake.file_name(extension='png') + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + # Count the number of files before attempting to add the file + cursor.execute("SELECT COUNT(*) FROM Content") + count_before = cursor.fetchone()[0] + # Attempt to add the file + self.db.add_file(png_path, jpeg_content, 'en-US') + # Verify the file is not added + cursor.execute("SELECT COUNT(*) FROM Content") + count_after = cursor.fetchone()[0] + self.assertEqual(count_after, count_before) + + def test_add_ttf_to_database(self): + # Create a simple TTF file + ttf_content = b'fake ttf content' + ttf_path = 'test.ttf' + with open(ttf_path, 'wb') as f: + f.write(ttf_content) + + # Add the TTF file to the database + self.db.add_file(ttf_path, ttf_content, 'en-US') + + # Check if the file was added + with self.db.get_connection() as connection: + cursor = connection.cursor() + cursor.execute("SELECT COUNT(*) FROM Content WHERE path = ?", (ttf_path,)) + count = cursor.fetchone()[0] + self.assertEqual(count, 1, "TTF file was not added to the database.") + + # Clean up + os.remove(ttf_path) + + def test_add_file_idempotency(self): + # Create a simple file + file_content = b'Hello, World!' + file_path = 'test_idempotency.txt' + with open(file_path, 'wb') as f: + f.write(file_content) + + # Add the file to the database + self.db.add_file(file_path, file_content, 'en-US') + + # Check if the file was added + with self.db.get_connection() as connection: + cursor = connection.cursor() + cursor.execute("SELECT COUNT(*) FROM Content WHERE path = ?", (file_path,)) + count_after_first = cursor.fetchone()[0] + self.assertEqual(count_after_first, 1, "File was not added to the database on the first call.") + + # Add the same file again + self.db.add_file(file_path, file_content, 'en-US') + + # Check if the file count remains unchanged + with self.db.get_connection() as connection: + cursor = connection.cursor() + cursor.execute("SELECT COUNT(*) FROM Content WHERE path = ?", (file_path,)) + count_after_second = cursor.fetchone()[0] + self.assertEqual(count_after_second, 1, "File count changed on the second call, violating idempotency.") + + # Clean up + os.remove(file_path) + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/scripts/test_create_empty_database.py b/scripts/test_create_empty_database.py new file mode 100644 index 000000000..fe3ac7194 --- /dev/null +++ b/scripts/test_create_empty_database.py @@ -0,0 +1,92 @@ +import unittest +import os +import tempfile +import sqlite3 +from DocumentationDatabase import DocumentationDatabase + +class TestDocumentationDatabase(unittest.TestCase): + def setUp(self): + self.temp_db_file = tempfile.NamedTemporaryFile(delete=False) + self.db = DocumentationDatabase(self.temp_db_file.name) + + def tearDown(self): + # Clean up the temporary database file if it exists + if os.path.exists(self.temp_db_file.name): + os.unlink(self.temp_db_file.name) + + def test_init_creates_database(self): + # Check if the database file exists + self.assertTrue(os.path.exists(self.temp_db_file.name)) + # Check if the tables are created + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") + tables = cursor.fetchall() + self.assertIn(('ide_tooltip_table',), tables) + self.assertIn(('Content',), tables) + self.assertIn(('Languages',), tables) + self.assertIn(('ContentTypes',), tables) + + def test_content_types_populated(self): + # Check if the ContentTypes table contains all expected types + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + cursor.execute("SELECT value FROM ContentTypes;") + content_types = {row[0] for row in cursor.fetchall()} + for mime_type in DocumentationDatabase.CONTENT_TYPES: + self.assertIn(mime_type, content_types) + + def test_languages_populated(self): + # Check if the Languages table contains 'en-US' + with sqlite3.connect(self.temp_db_file.name) as connection: + cursor = connection.cursor() + cursor.execute("SELECT value FROM Languages;") + languages = {row[0] for row in cursor.fetchall()} + self.assertIn('en-US', languages) + + def test_error_on_bad_schema(self): + # Create a DB with the wrong schema + with tempfile.NamedTemporaryFile(delete=False) as bad_db_file: + bad_db_file.close() + with sqlite3.connect(bad_db_file.name) as connection: + cursor = connection.cursor() + cursor.execute("CREATE TABLE WrongTable (id INTEGER PRIMARY KEY);") + # Should raise ValueError + with self.assertRaises(ValueError): + DocumentationDatabase(bad_db_file.name) + os.unlink(bad_db_file.name) + + def test_accept_valid_preexisting_database(self): + # Create a valid DB with the correct schema and content + with tempfile.NamedTemporaryFile(delete=False) as valid_db_file: + valid_db_file.close() + # Use the class's methods to create and populate the schema + db = DocumentationDatabase(valid_db_file.name) + # Should NOT raise ValueError + try: + with sqlite3.connect(valid_db_file.name) as connection: + cursor = connection.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") + tables = cursor.fetchall() + expected_tables = {'ide_tooltip_table', 'Content', 'Languages', 'ContentTypes'} + existing_tables = {table[0] for table in tables if not table[0].startswith('sqlite_')} + self.assertEqual(existing_tables, expected_tables) + # Instantiate DocumentationDatabase a second time to verify it accepts a preexisting valid database + db2 = DocumentationDatabase(valid_db_file.name) + finally: + os.unlink(valid_db_file.name) + + def test_database_file_persists_after_object_goes_out_of_scope(self): + # Create a temporary database file name + with tempfile.NamedTemporaryFile(delete=False) as temp_file: + temp_file.close() + os.unlink(temp_file.name) + # Create the database and let the object go out of scope + DocumentationDatabase(temp_file.name) + # Verify the file still exists + self.assertTrue(os.path.exists(temp_file.name)) + os.unlink(temp_file.name) + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/scripts/test_get_file.py b/scripts/test_get_file.py new file mode 100644 index 000000000..c62b92502 --- /dev/null +++ b/scripts/test_get_file.py @@ -0,0 +1,148 @@ +import unittest +from DocumentationDatabase import DocumentationDatabase +from faker import Faker +import tempfile +import os +from PIL import Image +import io + +class TestGetFile(unittest.TestCase): + def test_insert_and_retrieve_plain_text_file(self): + fake = Faker() + # Generate a random plain text file + text_content = fake.text().encode() + text_path = 'random_' + fake.file_name(extension='txt') + + # Create a temporary database file + temp_db_file = tempfile.NamedTemporaryFile(delete=False) + temp_db_file.close() + os.unlink(temp_db_file.name) + + # Initialize the database + db = DocumentationDatabase(temp_db_file.name) + + # Insert the file into the database + db.add_file(text_path, text_content, 'en-US') + + # Retrieve the file from the database + with db.get_file(text_path, 'en-US') as retrieved_file: + retrieved_content = retrieved_file.read() + + # Verify the retrieved content matches the original + self.assertEqual(retrieved_content, text_content) + + # Clean up the temporary database file + os.unlink(temp_db_file.name) + + def test_insert_and_retrieve_html_file(self): + fake = Faker() + # Generate a random HTML file + html_content = b'' + fake.sentence().encode() + b'' + fake.paragraph().encode() + b'' + html_path = 'random_' + fake.file_name(extension='html') + + # Create a temporary database file + temp_db_file = tempfile.NamedTemporaryFile(delete=False) + temp_db_file.close() + os.unlink(temp_db_file.name) + + # Initialize the database + db = DocumentationDatabase(temp_db_file.name) + + # Insert the file into the database + db.add_file(html_path, html_content, 'en-US') + + # Retrieve the file from the database + with db.get_file(html_path, 'en-US') as retrieved_file: + retrieved_content = retrieved_file.read() + + # Verify the retrieved content matches the original + self.assertEqual(retrieved_content, html_content) + + # Clean up the temporary database file + os.unlink(temp_db_file.name) + + def test_insert_and_retrieve_jpeg_file(self): + fake = Faker() + # Create a simple JPEG image using Pillow + img = Image.new('RGB', (800, 600), color='red') + img_byte_arr = io.BytesIO() + img.save(img_byte_arr, format='JPEG') + jpeg_content = img_byte_arr.getvalue() + jpeg_path = 'random_' + fake.file_name(extension='jpg') + + # Create a temporary database file + temp_db_file = tempfile.NamedTemporaryFile(delete=False) + temp_db_file.close() + os.unlink(temp_db_file.name) + + # Initialize the database + db = DocumentationDatabase(temp_db_file.name) + + # Insert the file into the database + db.add_file(jpeg_path, jpeg_content, 'en-US') + + # Retrieve the file from the database + with db.get_file(jpeg_path, 'en-US') as retrieved_file: + retrieved_content = retrieved_file.read() + + # Verify the retrieved content matches the original + self.assertEqual(retrieved_content, jpeg_content) + + # Clean up the temporary database file + os.unlink(temp_db_file.name) + + def test_insert_and_retrieve_png_file(self): + fake = Faker() + # Create a more interesting PNG image using Pillow + img = Image.new('RGB', (800, 600), color='green') + # Add a solid square in the middle + for x in range(300, 500): + for y in range(200, 400): + img.putpixel((x, y), (255, 0, 0)) # red + # Add some random colors + for x in range(0, 800, 100): + for y in range(0, 600, 100): + img.putpixel((x, y), (0, 0, 255)) # blue + for x in range(50, 800, 100): + for y in range(50, 600, 100): + img.putpixel((x, y), (255, 255, 0)) # yellow + img_byte_arr = io.BytesIO() + img.save(img_byte_arr, format='PNG') + png_content = img_byte_arr.getvalue() + png_path = 'random_' + fake.file_name(extension='png') + + # Create a temporary database file + temp_db_file = tempfile.NamedTemporaryFile(delete=False) + temp_db_file.close() + os.unlink(temp_db_file.name) + + # Initialize the database + db = DocumentationDatabase(temp_db_file.name) + + # Insert the file into the database + db.add_file(png_path, png_content, 'en-US') + + # Retrieve the file from the database + with db.get_file(png_path, 'en-US') as retrieved_file: + retrieved_content = retrieved_file.read() + + # Validate that the retrieved content is a valid PNG file + try: + Image.open(io.BytesIO(retrieved_content)) + except Exception as e: + self.fail(f"Retrieved content is not a valid PNG file: {e}") + + # Assert that the retrieved content is smaller than the original content + self.assertLess(len(retrieved_content), len(png_content), "PNG file was not compressed.") + + # Summarize the input size, output size, and percent saved + input_size = len(png_content) + output_size = len(retrieved_content) + percent_saved = ((input_size - output_size) / input_size) * 100 + print(f"PNG Test: Input size: {input_size} bytes, Output size: {output_size} bytes, Percent saved: {percent_saved:.2f}%") + + # Clean up the temporary database file + os.unlink(temp_db_file.name) + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/scripts/test_ingest.py b/scripts/test_ingest.py new file mode 100644 index 000000000..0274cba9c --- /dev/null +++ b/scripts/test_ingest.py @@ -0,0 +1,119 @@ +import unittest +from unittest.mock import patch, mock_open +import argparse +import tempfile +import os +import sqlite3 +from ingest import main, DocumentationDatabase + +class DummyPool: + def __enter__(self): return self + def __exit__(self, exc_type, exc_val, exc_tb): pass + def starmap(self, func, iterable): + return [func(*args) for args in iterable] + +class TestIngest(unittest.TestCase): + def setUp(self): + # Create a temporary database file for each test + self.temp_db = tempfile.NamedTemporaryFile(delete=False) + self.temp_db.close() + self.db_path = self.temp_db.name + + # Clean up any existing database + if os.path.exists(self.db_path): + os.unlink(self.db_path) + + # Initialize the database with the required schema + self.db = DocumentationDatabase(self.db_path) + + def tearDown(self): + # Clean up the temporary database file + os.unlink(self.db_path) + + @patch('argparse.ArgumentParser.parse_args') + @patch('builtins.open', new_callable=mock_open, read_data=b'Test content') + @patch('os.path.isfile', return_value=True) + @patch('ingest.DocumentationDatabase') + def test_single_file_ingestion(self, mock_db, mock_isfile, mock_file, mock_args): + mock_args.return_value = argparse.Namespace(file='test.txt', directory=None, path_to_db=self.db_path) + mock_db.return_value = self.db + self.db.emit_summary(label="test_single_file_ingestion before") + main() + self.db.emit_summary(label="test_single_file_ingestion after") + + @patch('multiprocessing.Pool', new=DummyPool) + @patch('argparse.ArgumentParser.parse_args') + @patch('os.listdir') + @patch('builtins.open') + @patch('os.path.isfile', return_value=True) + @patch('ingest.DocumentationDatabase') + def test_directory_ingestion(self, mock_db, mock_isfile, mock_file, mock_listdir, mock_args): + mock_args.return_value = argparse.Namespace(file=None, directory='test_dir', path_to_db=self.db_path) + mock_listdir.return_value = ['file1.txt', 'file2.txt', 'file3.md', 'image1.jpg', 'image2.png', 'image3.gif'] + mock_db.return_value = self.db + + # Create a mock file that returns different content based on the file being opened + def mock_file_side_effect(*args, **kwargs): + filename = args[0] + if isinstance(filename, int): + # Allow file descriptors to pass through to the real open + return open(filename, *args[1:], **kwargs) + if filename.endswith('.txt'): + return mock_open(read_data=b'Text content')(*args, **kwargs) + elif filename.endswith('.md'): + return mock_open(read_data=b'# Markdown content')(*args, **kwargs) + elif filename.endswith('.jpg'): + return mock_open(read_data=b'\xFF\xD8\xFF\xE0\x00\x10JFIF\x00\x01\x01\x01\x00H\x00H\x00\x00\xFF\xDB\x00C\x00')(*args, **kwargs) + elif filename.endswith('.png'): + return mock_open(read_data=b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82')(*args, **kwargs) + elif filename.endswith('.gif'): + return mock_open(read_data=b'GIF87a\x01\x00\x01\x00\x80\x00\x00\xff\xff\xff\x00\x00\x00!\xf9\x04\x01\x00\x00\x00\x00,\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02D\x01\x00;')(*args, **kwargs) + else: + return mock_open(read_data=b'Unknown content')(*args, **kwargs) + + mock_file.side_effect = mock_file_side_effect + + self.db.emit_summary(label="test_directory_ingestion before") + main() + self.db.emit_summary(label="test_directory_ingestion after") + + @patch('builtins.open', new_callable=mock_open, read_data=b'Test content') + @patch('os.path.isfile', return_value=True) + def test_relative_path_handling(self, mock_isfile, mock_file): + from ingest import process_file + + # Test various relative path patterns + test_cases = [ + ('../../a/b/c.txt', 'a/b/c.txt'), + ('../a/b/c.txt', 'a/b/c.txt'), + ('./a/b/c.txt', 'a/b/c.txt'), + ('a/b/c.txt', 'a/b/c.txt'), # No change needed + ('/a/b/c.txt', '/a/b/c.txt'), # Absolute path, no change + ] + + for input_path, expected_path in test_cases: + with self.subTest(input_path=input_path): + # Create a fresh database for each subtest + temp_db = tempfile.NamedTemporaryFile(delete=False) + temp_db.close() + db_path = temp_db.name + + try: + # Process the file + process_file(input_path, db_path) + + # Verify the path was normalized correctly in the database + with sqlite3.connect(db_path) as conn: + cursor = conn.cursor() + cursor.execute("SELECT path FROM Content WHERE path = ?", (expected_path,)) + result = cursor.fetchone() + self.assertIsNotNone(result, f"Path {expected_path} not found in database") + self.assertEqual(result[0], expected_path, + f"Path not normalized correctly. Expected {expected_path}, got {result[0]}") + finally: + # Clean up the temporary database + if os.path.exists(db_path): + os.unlink(db_path) + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/scripts/test_stats.py b/scripts/test_stats.py new file mode 100644 index 000000000..e41b85339 --- /dev/null +++ b/scripts/test_stats.py @@ -0,0 +1,83 @@ +import unittest +from DocumentationDatabase import DocumentationDatabase +import os +import sqlite3 + +class TestStats(unittest.TestCase): + def setUp(self): + self.db_path = '/tmp/stats-test.sqlite' + # Ensure a clean database for each test + if os.path.exists(self.db_path): + os.remove(self.db_path) + self.db = DocumentationDatabase(self.db_path) + + def tearDown(self): + if os.path.exists(self.db_path): + os.remove(self.db_path) + + def test_new_database_has_zero_files(self): + """Test that a new database contains zero files.""" + count, input_bytes, stored_bytes = self.db.stats() + self.assertEqual(count, 0) + self.assertEqual(input_bytes, 0) + self.assertEqual(stored_bytes, 0) + + def test_stats_after_inserting_text_file(self): + """Test that stats() returns correct values after inserting a text file.""" + # Create a larger text file with repeated content to ensure compression works + content = b'Hello, World! ' * 100 # Repeat the text 100 times + with open('/tmp/tiny.txt', 'wb') as f: + f.write(content) + with open('/tmp/tiny.txt', 'rb') as f: + content = f.read() + self.db.add_file('/tmp/tiny.txt', content, 'en-US') + count, input_bytes, stored_bytes = self.db.stats() + self.assertEqual(count, 1) + self.assertEqual(input_bytes, len(content)) + self.assertGreater(stored_bytes, 0) + self.assertLess(stored_bytes, input_bytes) # Verify compression worked + os.remove('/tmp/tiny.txt') + + def test_stats_after_inserting_1000_files(self): + """Test that stats() returns correct values after inserting 1000 files.""" + num_files = 1000 + content = b'Hello, World! ' * 10 # 140 bytes per file + total_input_bytes = 0 + for i in range(num_files): + file_path = f'/tmp/file_{i}.txt' + with open(file_path, 'wb') as f: + f.write(content) + with open(file_path, 'rb') as f: + file_content = f.read() + self.db.add_file(file_path, file_content, 'en-US') + total_input_bytes += len(file_content) + os.remove(file_path) + count, input_bytes, stored_bytes = self.db.stats() + self.assertEqual(count, num_files) + self.assertEqual(input_bytes, total_input_bytes) + self.assertGreater(stored_bytes, 0) + self.assertLess(stored_bytes, input_bytes) # Verify compression worked + + def test_stored_bytes_matches_database(self): + """Test that stored_bytes in the class matches the sum of bytes in the Content table.""" + num_files = 10 + content = b'Hello, World! ' * 10 + for i in range(num_files): + file_path = f'/tmp/file_db_{i}.txt' + with open(file_path, 'wb') as f: + f.write(content) + with open(file_path, 'rb') as f: + file_content = f.read() + self.db.add_file(file_path, file_content, 'en-US') + os.remove(file_path) + # Get stored_bytes from the class + _, _, stored_bytes_class = self.db.stats() + # Get sum of content lengths from the database + with sqlite3.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute("SELECT SUM(LENGTH(content)) FROM Content") + stored_bytes_db = cursor.fetchone()[0] + self.assertEqual(stored_bytes_class, stored_bytes_db) + +if __name__ == '__main__': + unittest.main() \ No newline at end of file