diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml new file mode 100644 index 0000000..bbf0609 --- /dev/null +++ b/.github/workflows/run-tests.yml @@ -0,0 +1,33 @@ +name: Run Tests + +on: + push: + branches: + - main + pull_request: + branches: + - main + +jobs: + run-tests: + runs-on: ubuntu-latest + + steps: + - name: Checkout repository + uses: actions/checkout@v3 + + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: '3.x' + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + python -m pip install -r test/requirements.txt + + - name: Make test.sh executable + run: chmod +x test.sh + + - name: Run test.sh + run: ./test.sh \ No newline at end of file diff --git a/.gitignore b/.gitignore index e0d8e0a..3c92728 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ dist .pypirc -struct \ No newline at end of file +struct +*.log +versions +.mypy_cache \ No newline at end of file diff --git a/README.md b/README.md index 838e580..73f64d9 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # kvProcessor -[**PYPI Package**](https://pypi.org/project/kvprocessor/) \ +[**PYPI Package**](https://pypi.org/project/kvprocessor/) **•** [**GitHub**](https://github.com/connor33341/kvProcessor) \ A Python package for processing and validating configuration dictionaries against a custom `.kv` file format. @@ -14,86 +14,212 @@ pip install kvprocessor ## File format +The `.kv` file format is a simple key-value configuration format with support for type validation and default values. Each line in a `.kv` file follows this syntax: + ```custom -# Comments are defined by a "#" -VARIBLENAME:DEFAULTVAULE +VARIABLENAME:DEFAULTVALUE ``` -## Usage +- **VARIABLENAME**: The name of the variable. +- **TYPE**: The expected type(s) of the variable. Multiple types can be separated by `|`. +- **DEFAULTVALUE**: The default value for the variable. Use `none` if no default value is provided. +- **Comments**: Both comments as a new line, or inline are supprorted, with the `#` character. + +### Example `.kv` file: +```custom +DATABASE_NAME:none +DATABASE_PORT:3306 +ENABLE_LOGGING:true +MAX_CONNECTIONS:none +``` + +## KV Manifests + +A KV manifest is a file that defines namespaces and their relationships. It is used to organize and manage configurations across multiple `.kv` files. Each line in a manifest file follows this syntax: + +```custom +namespace1:namespace2 +``` + +- **namespace1**: The namespace that dosent exist as a file, but has the value as **namespace2**. +- **namespace2**: The full namespace path + +### Example manifest file: +```custom +# A valid manifest +root:database +root:logging +database:connection +``` + +### Validating a manifest: +You can validate a manifest using the `KVManifestLoader`: + +```python +from kvprocessor.kvmanifestloader import KVManifestLoader + +manifest_path = "test/manifest.txt" +loader = KVManifestLoader(manifest_path) +loader.validate_manifest() # Validates the manifest structure +``` + +## Config.json + +The `config.json` file is used by the `KVStructLoader` to define the structure and metadata of the configuration. It includes details such as the version, root namespace, and manifest file. +### Example `config.json`: +**Note**: This example uses features from `0.1.10`, the current version is `0.2.14+`. Some extra parameters may be needed. +```json +{ + "version": "0.1.10", + "root": "root", + "manifest": "manifest.txt", + "platform": "github", + "owner": "Voxa-Communications", + "repo": "VoxaCommunicaitons-Structures", + "branch": "main" +} +``` + +### Using `KVStructLoader` with `config.json`: ```python -from kvprocessor import LoadEnv, KVProcessor +from kvprocessor import KVStructLoader + +kv_config_url = "https://github.com/Voxa-Communications/VoxaCommunicaitons-Structures/raw/refs/heads/main/struct/config.json" +kv_struct_loader = KVStructLoader(kv_config_url) +kv_processor = kv_struct_loader.from_namespace("root.database.connection") +``` + +## Usage -kv_file_path = "test/test.kv" # Directory to .kv file -kv_processor = KVProcessor(kv_file_path) # Create a KV processor class -kv_keys = kv_processor.return_names() # Gets the keys (VARIBLENAME) from the .kv file -env_list = LoadEnv(kv_keys) # Loads all the ENV varibles that match those keys -validated_config = kv_processor.process_config(env_list) # Verifies that those env varibles exist and are of the correct type +### KVProcessor +```python +from kvprocessor import KVProcessor +from kvprocessor.kvenvloader import load_env + +kv_file_path = "test/test.kv" # Directory to .kv file +kv_processor = KVProcessor(kv_file_path) # Create a KV processor class +kv_keys = kv_processor.return_names() # Gets the keys (VARIBLENAME) from the .kv file +env_list = load_env(kv_keys) # Loads all the ENV variables that match those keys +validated_config = kv_processor.process_config(env_list) # Verifies that those env variables exist and are of the correct type print(validated_config) ``` -This example mimics the one found in the `/test` directory. With the kv file of: -```custom -# This is a comment -DATABASE_NAME:none -DATABASE_USER:none -DATABASE_PASSWORD:none -DATABASE_HOST:none -DATABASE_PORT:none -DATABASE_DRIVER:mysql+mysqlconnector -DATABASE_DIALECT:none -``` -You **should** get a result of: -`{'DATABASE_NAME': None, 'DATABASE_USER': None, 'DATABASE_PASSWORD': None, 'DATABASE_HOST': None, 'DATABASE_PORT': None, 'DATABASE_DRIVER': None, 'DATABASE_DIALECT': None}` This is because the kvProcessor is taking input from the env, and we dont have these env varibles defined. As a result these values default to the defined default value - -### Using "Namespaces" -This allows you to "import" kv files from a static host. +### KVStructLoader ```python -from kvprocessor import KVProcessor, KVStructLoader +from kvprocessor import KVStructLoader -kv_config_url = "https://github.com/Voxa-Communications/VoxaCommunicaitons-Structures/raw/refs/heads/main/struct/config.json" # STATIC url to json config -kv_struct_loader = KVStructLoader(kv_config_url) # Create a KVStructLoader object with the URL of the config file -kv_processor: KVProcessor = kv_struct_loader.from_namespace("voxa.api.user.user_settings") # Loads the KV file from the URL and returns a KVProcessor object -# For example this loads a file in /api/user/user_settings.kv +kv_config_url = "https://github.com/Voxa-Communications/VoxaCommunicaitons-Structures/raw/refs/heads/main/struct/config.json" +kv_struct_loader = KVStructLoader(kv_config_url) +kv_processor = kv_struct_loader.from_namespace("root.database.connection") user_settings = { - "2FA_ENABLED": True, - "TELEMETRY": False, - "AGE": "25", - "LANGUAGE": "en", -} # Example Dict Structure + "DATABASE_NAME": "test_db", + "DATABASE_PORT": 5432, +} validated_config = kv_processor.process_config(user_settings) print(validated_config) ``` -For an example config.json navigate to `test/config.json`. This file is just what is found on `https://github.com/Voxa-Communications/VoxaCommunicaitons-Structures/blob/main/struct/config.json` which is used in this example. -### Namespace's config.json -Namespace JSON files, have to be on a static host. They cannot be used locally. The easiest way to do this is to make a github repo, and use the raw file. -#### A config.json in a namespace should include at least two parts: - - A "root", the name that preceedes the rest of the namespace. Ex: `voxa` in `voxa.api.user.user_settings` - - A "URL". Ex: `https://mysite.example/kvstructures`, when the namespace `mysite.folder.structure` is used (assuming `root` is set to `mysite`), will fetch `https://mysite.example/kvstructures/folder/structure.kv` +### KVFileMerger +```python +from kvprocessor import KVFileMerger - Here is an example JSON (Note: on 0.7.1+ the URL is not needed, however a `struct` needs to be defined): -```json -{ - "root": "voxa", - "version": "0.1.5", - "URL": "https://raw.githubusercontent.com/Voxa-Communications/VoxaCommunicaitons-Structures/refs/heads/main/struct/" +file1 = "test/file1.kv" +file2 = "test/file2.kv" +merger = KVFileMerger(file1, file2) +merged_file = merger.merge("merged.kv") # Merges two KV files into a new file +print(f"Merged file created at: {merged_file}") +``` + +### KVFileUtils +```python +from kvprocessor.kvfileutils import search_kv_files, copy_kv_file, delete_kv_file + +# Search for KV files in a directory +kv_files = search_kv_files("test") +print(f"Found KV files: {kv_files}") + +# Copy a KV file +copy_kv_file("test/test.kv", "test/copy_test.kv") +print("KV file copied.") + +# Delete a KV file +delete_kv_file("test/copy_test.kv") +print("KV file deleted.") +``` + +### KVFileDiffChecker +```python +from kvprocessor import KVFileDiffChecker + +file1 = "test/file1.kv" +file2 = "test/file2.kv" +diff_checker = KVFileDiffChecker(file1, file2) +differences = diff_checker.diff() +print(f"Differences between files: {differences}") +``` + +### KVValidator +```python +from kvprocessor import validate_kv_file + +kv_file_path = "test/test.kv" +is_valid = validate_kv_file(kv_file_path) +print(f"KV file is valid: {is_valid}") +``` + +### Additional Data Types +The library supports additional data types such as `datetime`, `date`, `time`, and `decimal`. These can be used in `.kv` files as follows: + +```custom +EVENT_DATE:none +PRICE:none +``` + +Example usage: +```python +from kvprocessor import KVProcessor + +kv_file_path = "test/test.kv" # Path to your .kv file +kv_processor = KVProcessor(kv_file_path) + +# Example configuration with additional data types +config = { + "EVENT_DATE": "2025-05-01T12:00:00", + "PRICE": "19.99", } + +validated_config = kv_processor.process_config(config) +print(validated_config) ``` ## Building For building the library locally \ -**Requires**: `python3.8+`, `pip`, `linux system`(if using the predefined shell files) +**Requires**: `python3.8+`, `pip`, `linux system` (if using the predefined shell files) + +1. `git clone https://github.com/connor33341/kvProcessor.git` +2. `cd kvProcessor` +3. `bash build.sh` + +`build.sh` will also install kvProcessor as a local package, which you will be able to use. If you add new features to your fork and would like them to be featured on the main repo, submit a Pull Request. + +## CLI +At the current moment, there exists no documentation on this. If you would like to find usage, visit the file `kvprocessor\cli.py`. \ +\ +Basic Usage: +```bash +python kvprocessor/cli.py --version +``` - 1. `git clone https://github.com/connor33341/kvProcessor.git` - 2. `cd kvProcessor` - 3. `bash build.sh` +## Library Modules +For a complete list, visit `kvprocessor\__init__.py`. A breif list of main modules, will be listed here. + - `kvprocessor.kvprocessor`, Exports: `KVProcessor` + - `kvprocessor.kvstructloader`, Exports: `KVStructLoader` + - `kvprocessor.kvmanifestloader`, Exports: `KVManifestLoader` -`build.sh` will also install kvProcessor as a local package, which you will be able to use. -If you add new features to your fork, and would like them to be featured on the main repo. Submit a Pull Request ## For the nerds -The syntax was already mentioned, however if you would like to see how it parses. The following regex is used to determine the: `name`, `type`, and `default`: +The syntax was already mentioned, however, if you would like to see how it parses, the following regex is used to determine the: `name`, `type`, and `default`: ```re (\w+)<([\w\|]+)>:([\w+]+|none) ``` -With this knowlege, you probably can figure out a way to write .kv files in a weird way. Out of typical standard. \ No newline at end of file +With this knowledge, you probably can figure out a way to write `.kv` files in a weird way, out of typical standard. \ No newline at end of file diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..e583b1e --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,20 @@ +# Security Policy + +## Supported Versions + +Versions below `0.1.7` are no longer supproted. `0.2.12`, and `0.2.14-alpha` are both not supproted. Migrate to `0.2.14` w/o the alpha designation, for a stable release. + +| Version | Supported | +| ------- | ------------------ | +| 0.2.14+ | :white_check_mark: | +| 0.2.12 | :x: | +| 0.2.1-2 | :x: | +| 0.1.10+ | :white_check_mark: | +| < 0.1.7 | :x: | + +## Reporting a Vulnerability + +Contact the developer: \ +\ +Vunerabilites are rare, on this library. However, there is still a potental for them to exist. Please report these, so that they can be delt with accordingly. +There is no set style inwhich you have to submit a Vunerability. However, I would recomend being detailed, and having the information of what file or library is causing the issue. diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..8b28231 --- /dev/null +++ b/TODO.md @@ -0,0 +1,17 @@ +# TODO + +### Documentation: + - Create a doc/wiki with mkdocs + - Update `README.md` to have basic examples for 0.2.x versions of the API + +### Testing: + - Update test.py to test for most usecases with the 0.2.x API + - Use pytest aswell + - Create a GHAction to makesure that all tests are passed. + +### API: + - Migrate to `kvprocessor.errors` for custom errors + - Update cli to support 0.2.x verions of the API + - Expand `kvprocessor.kvtypemap` to support more types. + - Update `kvprocessor.kvmanifestloader` to be more feature rich + - Fix the style, so it will lint correctly \ No newline at end of file diff --git a/kvprocessor/__init__.py b/kvprocessor/__init__.py index 409e481..c88edd1 100644 --- a/kvprocessor/__init__.py +++ b/kvprocessor/__init__.py @@ -1,6 +1,26 @@ -__version__ = "0.1.12" +__version__ = "0.2.14" from .kvprocessor import KVProcessor -from .kvenvloader import LoadEnv +from .kvenvloader import load_env, LoadEnv from .kvstructloader import KVStructLoader -from .errors import KVProcessorError, InvalidKVFileError, MissingEnvironmentVariableError, NamespaceNotFoundError, InvalidNamespaceError \ No newline at end of file +from .kvfileexporter import KVFileExporter +from .kvfilemerger import KVFileMerger +from .kvfileutils import ( + search_kv_files, + copy_kv_file, + delete_kv_file, +) +from .kvdiff import KVFileDiffChecker +from .kvnamespacemanager import NamespaceManager as KVNamespaceManager +from .kvmanifestloader import KVManifestLoader +from .kvversionmanager import KVVersionManager +from .kvdiff import KVFileDiffChecker +from .kvvalidator import KVFileValidator, validate_kv_file, validate_kv_key, validate_kv_value +from .kvtypemap import get_type_map, set_type_map, remove_type_map, has_type_map, clear_type_map, add_type_map +from .kvglobalsettings import set_version, get_version, get_version_tuple, get_version_major, get_version_minor +from .util.errors import KVProcessorError, InvalidKVFileError, MissingEnvironmentVariableError, NamespaceNotFoundError, InvalidNamespaceError +from .util.warnings import deprecated as kv_deprecated_warning +from .util.log import log as kv_log + +# CLI +from .cli import main as kv_cli_main # idealy you dont use this, and use cli directly, however, if you are lazy, just import it, then run kvprocessor.kv_cli_main() to run the CLI \ No newline at end of file diff --git a/kvprocessor/cli.py b/kvprocessor/cli.py new file mode 100644 index 0000000..71e98c8 --- /dev/null +++ b/kvprocessor/cli.py @@ -0,0 +1,181 @@ +import argparse +from kvprocessor.kvvalidator import validate_kv_file +from kvprocessor.kvmanifestloader import KVManifestLoader +from kvprocessor.kvnamespacemanager import NamespaceManager +from kvprocessor.kvprocessor import KVProcessor +from kvprocessor.kvglobalsettings import get_version + +def main(): + parser = argparse.ArgumentParser(description="CLI for kvProcessor") + parser.add_argument("--version", action="version", version=f"kvProcessor {get_version()}", help="Show the version of kvProcessor") + subparsers = parser.add_subparsers(dest="command") + + # Subcommand: Validate .kv file + validate_parser = subparsers.add_parser("validate", help="Validate a .kv file") + validate_parser.add_argument("file", type=str, help="Path to the .kv file") + + # Subcommand: List namespaces + list_parser = subparsers.add_parser("list-namespaces", help="List all namespaces") + list_parser.add_argument("manifest", type=str, help="Path to the manifest file") + + # Subcommand: Validate manifest + validate_manifest_parser = subparsers.add_parser("validate-manifest", help="Validate a manifest file") + validate_manifest_parser.add_argument("manifest", type=str, help="Path to the manifest file") + + # Subcommand: Add namespace + add_parser = subparsers.add_parser("add-namespace", help="Add a new namespace") + add_parser.add_argument("manifest", type=str, help="Path to the manifest file") + add_parser.add_argument("key", type=str, help="Namespace key") + add_parser.add_argument("value", type=str, help="Namespace value") + + # Subcommand: Remove namespace + remove_parser = subparsers.add_parser("remove-namespace", help="Remove a namespace") + remove_parser.add_argument("manifest", type=str, help="Path to the manifest file") + remove_parser.add_argument("key", type=str, help="Namespace key") + + # Subcommand: Export configuration + export_parser = subparsers.add_parser("export-config", help="Export configuration to a .kv file") + export_parser.add_argument("config", type=str, help="Path to the configuration JSON file") + export_parser.add_argument("output", type=str, help="Path to the output .kv file") + + # Subcommand: Merge .kv files + merge_parser = subparsers.add_parser("merge-kv", help="Merge multiple .kv files into one") + merge_parser.add_argument("files", nargs='+', help="Paths to the .kv files to merge") + merge_parser.add_argument("output", type=str, help="Path to the output .kv file") + + # Subcommand: Diff .kv files + diff_parser = subparsers.add_parser("diff-kv", help="Show differences between two .kv files") + diff_parser.add_argument("file1", type=str, help="Path to the first .kv file") + diff_parser.add_argument("file2", type=str, help="Path to the second .kv file") + + # Subcommand: Generate namespace report + report_parser = subparsers.add_parser("namespace-report", help="Generate a report of namespaces") + report_parser.add_argument("manifest", type=str, help="Path to the manifest file") + report_parser.add_argument("output", type=str, help="Path to the output report file") + + # Subcommand: Convert legacy config + convert_parser = subparsers.add_parser("convert-legacy", help="Convert legacy configuration to new format") + convert_parser.add_argument("legacy_config", type=str, help="Path to the legacy configuration file") + convert_parser.add_argument("output", type=str, help="Path to the output configuration file") + + # Subcommand: Backup .kv file + backup_parser = subparsers.add_parser("backup-kv", help="Create a backup of a .kv file") + backup_parser.add_argument("file", type=str, help="Path to the .kv file") + backup_parser.add_argument("backup_dir", type=str, help="Directory to store the backup") + + args = parser.parse_args() + + if args.command == "validate": + try: + if validate_kv_file(args.file): + print(f"{args.file} is valid.") + except Exception as e: + print(f"Validation failed: {e}") + + elif args.command == "list-namespaces": + try: + manifest_loader = KVManifestLoader(args.manifest) + manager = NamespaceManager(manifest_loader) + namespaces = manager.list_namespaces() + print("Available namespaces:") + for namespace in namespaces: + print(namespace) + except Exception as e: + print(f"Error listing namespaces: {e}") + + elif args.command == "validate-manifest": + try: + manifest_loader = KVManifestLoader(args.manifest) + manifest_loader.validate_manifest() + print(f"Manifest {args.manifest} is valid.") + except Exception as e: + print(f"Manifest validation failed: {e}") + + elif args.command == "add-namespace": + try: + manifest_loader = KVManifestLoader(args.manifest) + manager = NamespaceManager(manifest_loader) + manager.add_namespace(args.key, args.value) + print(f"Namespace {args.key} added successfully.") + except Exception as e: + print(f"Error adding namespace: {e}") + + elif args.command == "remove-namespace": + try: + manifest_loader = KVManifestLoader(args.manifest) + manager = NamespaceManager(manifest_loader) + manager.remove_namespace(args.key) + print(f"Namespace {args.key} removed successfully.") + except Exception as e: + print(f"Error removing namespace: {e}") + + elif args.command == "export-config": + from kvprocessor.kvfileexporter import KVFileExporter + import json + try: + with open(args.config, 'r') as config_file: + config = json.load(config_file) + exporter = KVFileExporter(args.output) + exporter.validate_and_export(config) + print(f"Configuration exported to {args.output}.") + except Exception as e: + print(f"Error exporting configuration: {e}") + + elif args.command == "merge-kv": + from kvprocessor.kvfilemerger import KVFileMerger + try: + merger = KVFileMerger(args.output) + merger.merge(args.files) + print(f"Merged .kv files into {args.output}.") + except Exception as e: + print(f"Error merging .kv files: {e}") + + elif args.command == "diff-kv": + from kvprocessor.kvdiff import KVFileDiffChecker as KVFileDiff + try: + differ = KVFileDiff(args.file1, args.file2) + differences = differ.diff() + print("Differences between files:") + print(differences) + except Exception as e: + print(f"Error diffing .kv files: {e}") + + elif args.command == "namespace-report": + try: + manifest_loader = KVManifestLoader(args.manifest) + manager = NamespaceManager(manifest_loader) + namespaces = manager.list_namespaces() + with open(args.output, 'w') as report_file: + report_file.write("Namespace Report\n") + report_file.write("================\n") + for namespace in namespaces: + report_file.write(f"{namespace}\n") + print(f"Namespace report generated at {args.output}.") + except Exception as e: + print(f"Error generating namespace report: {e}") + + elif args.command == "convert-legacy": + from kvprocessor.kvstructloader import KVStructLoader + try: + struct_loader = KVStructLoader(args.legacy_config) + new_config = struct_loader.convert_to_new_format() + with open(args.output, 'w') as output_file: + output_file.write(new_config) + print(f"Legacy configuration converted and saved to {args.output}.") + except Exception as e: + print(f"Error converting legacy configuration: {e}") + + elif args.command == "backup-kv": + import os + import shutil + try: + if not os.path.exists(args.backup_dir): + os.makedirs(args.backup_dir) + backup_path = os.path.join(args.backup_dir, os.path.basename(args.file)) + shutil.copy2(args.file, backup_path) + print(f"Backup created at {backup_path}.") + except Exception as e: + print(f"Error creating backup: {e}") + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/kvprocessor/kvdiff.py b/kvprocessor/kvdiff.py new file mode 100644 index 0000000..ad3f9d2 --- /dev/null +++ b/kvprocessor/kvdiff.py @@ -0,0 +1,18 @@ +class KVFileDiffChecker: + """Compares two .kv files and highlights differences.""" + def __init__(self, file1: str, file2: str): + self.file1 = file1 + self.file2 = file2 + + def diff(self) -> dict: + with open(self.file1, 'r') as f1, open(self.file2, 'r') as f2: + lines1 = {line.strip() for line in f1 if line.strip() and not line.startswith('#')} + lines2 = {line.strip() for line in f2 if line.strip() and not line.startswith('#')} + + added = lines2 - lines1 + removed = lines1 - lines2 + + return { + 'added': added, + 'removed': removed + } \ No newline at end of file diff --git a/kvprocessor/kvenvloader.py b/kvprocessor/kvenvloader.py index 38297f4..d8e964c 100644 --- a/kvprocessor/kvenvloader.py +++ b/kvprocessor/kvenvloader.py @@ -1,8 +1,53 @@ import os +import warnings +from typing import Any, Optional +from kvprocessor.util.warnings import deprecated, ignore_warnings -def LoadEnv(Names: list) -> dict[str, any]: +def load_env(Names: list, defaults: Optional[dict] = None) -> dict[str, Any]: + """ + Load environment variables into a dictionary. + + Args: + Names (list): List of environment variable names to load. + defaults (dict): Optional dictionary of default values for environment variables. + + Returns: + dict: A dictionary containing the environment variable names and their values. + """ EnvList = {} + defaults = defaults or {} for Name in Names: - Value = os.environ.get(Name) + Value = os.environ.get(Name, defaults.get(Name)) + if Value is None: + warnings.warn(f"Environment variable '{Name}' is not set and no default value is provided.", UserWarning) EnvList[Name] = Value return EnvList + +def validate_env(Names: list) -> None: + """ + Validate that all required environment variables are set. + + Args: + Names (list): List of environment variable names to validate. + + Raises: + EnvironmentError: If any required environment variable is not set. + """ + missing_vars = [Name for Name in Names if os.environ.get(Name) is None] + if missing_vars: + raise EnvironmentError(f"The following required environment variables are missing: {', '.join(missing_vars)}") + +@ignore_warnings +@deprecated +def LoadEnv(Names: list, defaults: Optional[dict] = None) -> dict[str, Any]: + """ + Deprecated wrapper for load_env. + + Args: + Names (list): List of environment variable names to load. + defaults (dict): Optional dictionary of default values for environment variables. + + Returns: + dict: A dictionary containing the environment variable names and their values. + """ + return load_env(Names, defaults) \ No newline at end of file diff --git a/kvprocessor/kvfileexporter.py b/kvprocessor/kvfileexporter.py new file mode 100644 index 0000000..a787bd5 --- /dev/null +++ b/kvprocessor/kvfileexporter.py @@ -0,0 +1,39 @@ +class KVFileExporter: + """Exports processed configuration back into a .kv file format.""" + def __init__(self, output_path: str): + self.output_path = output_path + + def export(self, config: dict): + """Exports the configuration dictionary to a .kv file.""" + try: + with open(self.output_path, 'w') as file: + for key, value in config.items(): + value_type = type(value).__name__ + if value is None: + value = 'none' + file.write(f"{key}<{value_type}>:{value}\n") + except IOError as e: + raise IOError(f"Failed to write to file {self.output_path}: {e}") + + def validate_and_export(self, config: dict): + """Validates the configuration dictionary before exporting.""" + if not isinstance(config, dict): + raise TypeError("Configuration must be a dictionary.") + + for key, value in config.items(): + if not isinstance(key, str): + raise ValueError(f"Invalid key type: {key}. Keys must be strings.") + + self.export(config) + + def append_to_file(self, additional_config: dict): + """Appends additional configuration to the existing .kv file.""" + try: + with open(self.output_path, 'a') as file: + for key, value in additional_config.items(): + value_type = type(value).__name__ + if value is None: + value = 'none' + file.write(f"{key}<{value_type}>:{value}\n") + except IOError as e: + raise IOError(f"Failed to append to file {self.output_path}: {e}") \ No newline at end of file diff --git a/kvprocessor/kvfilemerger.py b/kvprocessor/kvfilemerger.py new file mode 100644 index 0000000..cbf3731 --- /dev/null +++ b/kvprocessor/kvfilemerger.py @@ -0,0 +1,18 @@ +class KVFileMerger: + """Merges multiple .kv files into one.""" + def __init__(self, output_path: str): + self.output_path = output_path + + def merge(self, file_paths: list): + merged_data = {} + for file_path in file_paths: + with open(file_path, 'r') as file: + for line in file: + line = line.strip() + if not line or line.startswith('#'): + continue + key, _, value = line.partition(':') + merged_data[key] = value + with open(self.output_path, 'w') as file: + for key, value in merged_data.items(): + file.write(f"{key}:{value}\n") \ No newline at end of file diff --git a/kvprocessor/kvfileutils.py b/kvprocessor/kvfileutils.py new file mode 100644 index 0000000..5b3924f --- /dev/null +++ b/kvprocessor/kvfileutils.py @@ -0,0 +1,23 @@ +import os +import shutil + +def search_kv_files(directory: str) -> list: + """Search for all .kv files in a directory.""" + kv_files = [] + for root, _, files in os.walk(directory): + for file in files: + if file.endswith(".kv"): + kv_files.append(os.path.join(root, file)) + return kv_files + +def copy_kv_file(source: str, destination: str): + """Copy a .kv file to a new location.""" + if not os.path.exists(source): + raise FileNotFoundError(f"Source file not found: {source}") + shutil.copy(source, destination) + +def delete_kv_file(file_path: str): + """Delete a .kv file.""" + if not os.path.exists(file_path): + raise FileNotFoundError(f"File not found: {file_path}") + os.remove(file_path) \ No newline at end of file diff --git a/kvprocessor/kvglobalsettings.py b/kvprocessor/kvglobalsettings.py new file mode 100644 index 0000000..ebeafc5 --- /dev/null +++ b/kvprocessor/kvglobalsettings.py @@ -0,0 +1,28 @@ +import kvprocessor +from urllib.parse import urlparse, urlunparse + +version = str(kvprocessor.__version__) + +def set_version(v: str): + """Set the version of the KVProcessor.""" + global version + version = v + +def get_version() -> str: + """Get the version of the KVProcessor.""" + return version +def get_version_tuple() -> tuple[int, ...]: + """Get the version of the KVProcessor as a tuple.""" + version_parts = version.split('.') + return tuple(map(int, version_parts[:3])) +def get_version_major() -> int: + """Get the major version of the KVProcessor.""" + return int(version.split('.')[0]) +def get_version_minor() -> int: + """Get the minor version of the KVProcessor.""" + return int(version.split('.')[1]) +@DeprecationWarning +def get_config_version_from_url(url: str) -> str: + """Get the config version of the from a URL.""" + from kvprocessor.kvstructloader import KVStructLoader # Local import to avoid circular dependency + return KVStructLoader(str(urlparse(url).path.rsplit('/', 1)[0] + '/config.json')).version \ No newline at end of file diff --git a/kvprocessor/kvmanifestloader.py b/kvprocessor/kvmanifestloader.py index bb71876..39b27e9 100644 --- a/kvprocessor/kvmanifestloader.py +++ b/kvprocessor/kvmanifestloader.py @@ -1,18 +1,27 @@ import os import requests import re +from urllib.parse import urlparse, urlunparse +from kvprocessor.kvglobalsettings import get_version_major, get_version_minor, get_version from kvprocessor.kvprocessor import KVProcessor -from kvprocessor.log import log +from kvprocessor.util.log import log +from kvprocessor.util.warnings import ignore_warnings +from kvprocessor.util.errors import ManifestError +from kvprocessor.util.filedownloader import download_file +from typing import Optional class KVManifestLoader: - def __init__(self, file_url: str, cache_dir: str = "./struct", root: str = None): + def __init__(self, file_url: str, cache_dir: str = "./struct", root: Optional[str] = None, manifest_version: str = get_version()): self.file_url = file_url self.cache_dir = cache_dir + self.manifest_version = manifest_version self.root = root self.manifest = None - self.namespace_overides = {} + self.namespace_overrides: dict[str, str] = {} self._fetch_manifest() self._parse_manifest() + if int(str(self.manifest_version).strip().split(".")[1]) >= 2: + self.validate_manifest() def _fetch_manifest(self): try: @@ -30,10 +39,12 @@ def _fetch_manifest(self): except requests.RequestException as e: print(f"Error fetching manifest file: {e}") return None - + + @ignore_warnings def _parse_manifest(self): try: - with open(os.path.join(self.cache_dir, f"{self.root}.txt"), 'r') as file: + manifest_path = os.path.join(self.cache_dir, f"{self.root}.txt") + with open(manifest_path, 'r') as file: self.manifest = file.read() log(f"Manifest loaded: {self.manifest}") i = -1 @@ -42,18 +53,56 @@ def _parse_manifest(self): line = line.strip() if not line or line.startswith('#'): continue - match: dict = re.match(r'([^:]+):([^:]+)', line) + if line.startswith('$import'): + # Handle $import directive + imported_file = line.split(' ', 1)[1].strip() + imported_path = os.path.join(self.cache_dir, imported_file) + + if not os.path.exists(imported_path): + # Attempt to fetch the file from the URL + log(f"Imported manifest file not found locally: {imported_path}") + manifest_url = urlparse(self.file_url)._replace(path=f"/{imported_file}").geturl() + log(f"Attempting to fetch imported manifest from URL: {manifest_url}") + + try: + download_file(manifest_url, imported_path) + log(f"Successfully fetched and saved imported manifest: {imported_path}") + except requests.RequestException as e: + raise FileNotFoundError(f"Failed to fetch imported manifest from URL: {manifest_url}. Error: {e}") + + # Read the imported manifest + with open(imported_path, 'r') as imported_file: + imported_content = imported_file.read() + # Append imported content with a separating comment + self.manifest += f"\n# Imported from {imported_file.name}\n{imported_content}" + continue + + match = re.match(r'([^:]+):([^:]+)', line) if not match: - if (len(line.split(":")) == 0) and (len(line.split(".") >= 1)): + if (len(line.split(":")) == 0) and (len(line.split(".")) >= 1): log("Found namespace") match.clear() match[str(line).strip()] = str(line).strip() raise ValueError(f"Invalid manifest file format in line: {line}") else: - log("Found namespace overide") + log("Found namespace override") key, value = match.groups() - log(f"Parsing Line {i} key={key}, value={value}") - self.namespace_overides[key] = value + log(f"Parsing Line {i} key={key}, value={value}") + self.namespace_overrides[key] = value except FileNotFoundError: - print(f"Manifest file not found: {self.file_url}") - return None \ No newline at end of file + print(f"Manifest file not found: {manifest_path}") + return None + + def validate_manifest(self): + """Validates the manifest for required fields and structure.""" + if not self.manifest: + raise ManifestError("Manifest is not loaded.") + + for i, line in enumerate(self.manifest.splitlines(), start=1): + line = line.strip() + if not line or line.startswith('#'): + continue + if ':' not in line: + raise ManifestError(f"Invalid manifest format at line {i}: {line}") + + log("Manifest validation passed.") \ No newline at end of file diff --git a/kvprocessor/kvnamespacemanager.py b/kvprocessor/kvnamespacemanager.py new file mode 100644 index 0000000..efda70b --- /dev/null +++ b/kvprocessor/kvnamespacemanager.py @@ -0,0 +1,29 @@ +from kvprocessor.kvmanifestloader import KVManifestLoader + +class NamespaceManager: + """Utility class for managing namespaces dynamically.""" + + def __init__(self, manifest_loader: KVManifestLoader): + self.manifest_loader = manifest_loader + + def add_namespace(self, key: str, value: str): + """Add a new namespace to the manifest.""" + if key in self.manifest_loader.namespace_overrides: + raise ValueError(f"Namespace {key} already exists.") + self.manifest_loader.namespace_overrides[key] = value + + def remove_namespace(self, key: str): + """Remove a namespace from the manifest.""" + if key not in self.manifest_loader.namespace_overrides: + raise KeyError(f"Namespace {key} does not exist.") + del self.manifest_loader.namespace_overrides[key] + + def list_namespaces(self) -> list: + """List all available namespaces.""" + return list(self.manifest_loader.namespace_overrides.keys()) + + def update_namespace(self, key: str, new_value: str): + """Update an existing namespace.""" + if key not in self.manifest_loader.namespace_overrides: + raise KeyError(f"Namespace {key} does not exist.") + self.manifest_loader.namespace_overrides[key] = new_value \ No newline at end of file diff --git a/kvprocessor/kvprocessor.py b/kvprocessor/kvprocessor.py index e8e7141..1c0c0a8 100644 --- a/kvprocessor/kvprocessor.py +++ b/kvprocessor/kvprocessor.py @@ -1,7 +1,8 @@ import re from typing import Dict, Any, Union -from kvprocessor.log import log -from kvprocessor.errors import InvalidKVFileError +from kvprocessor.util.log import log +from kvprocessor.util.errors import InvalidKVFileError +from kvprocessor.kvtypemap import get_type_map class KVProcessor: def __init__(self, kv_file_path: str): @@ -36,24 +37,12 @@ def _parse_kv_file(self, file_path: str) -> Dict[str, dict]: def _validate_type(self, value: Any, expected_types: list) -> bool: """Validate if the value matches one of the expected types.""" - type_map = { - 'string': str, - 'int': int, - 'float': float, - 'bool': bool, - 'none': type(None), - 'list': list, - 'dict': dict, - 'tuple': tuple, - 'set': set, - 'object': object, - 'any': Any, - 'str': str - } + type_map = get_type_map() for type_name in expected_types: if type_name not in type_map: raise ValueError(f"Unsupported type in .kv file: {type_name}") - if isinstance(value, type_map[type_name]): + expected_type = type_map[type_name] if isinstance(type_map[type_name], tuple) else (type_map[type_name],) + if isinstance(value, expected_type): return True return False @@ -87,4 +76,4 @@ def process_config(self, config: Dict[str, Any]) -> Dict[str, Any]: def return_names(self) -> list: """Return the names of the keys in the KV file.""" - return list(self.config_spec.keys()) \ No newline at end of file + return list(self.config_spec.keys()) diff --git a/kvprocessor/kvstructloader.py b/kvprocessor/kvstructloader.py index b66c421..8a298ed 100644 --- a/kvprocessor/kvstructloader.py +++ b/kvprocessor/kvstructloader.py @@ -4,24 +4,9 @@ from pathlib import Path from kvprocessor.kvprocessor import KVProcessor from kvprocessor.kvmanifestloader import KVManifestLoader -from kvprocessor.log import log -from kvprocessor.errors import NamespaceNotFoundError, InvalidNamespaceError - -class KVStructLoaderError(Exception): - """Base exception for KVStructLoader.""" - pass - -class ConfigFetchError(KVStructLoaderError): - """Raised when there is an error fetching the configuration.""" - pass - -class KVFetchError(KVStructLoaderError): - """Raised when there is an error fetching a KV file.""" - pass - -class ManifestError(KVStructLoaderError): - """Raised when there is an issue with the manifest.""" - pass +from kvprocessor.util.log import log +from kvprocessor.util.errors import NamespaceNotFoundError, InvalidNamespaceError, ManifestError, ConfigFetchError, KVFetchError +from kvprocessor.kvglobalsettings import get_version class KVStructLoader: def __init__(self, config_file: str, cache_dir: str = "./struct"): @@ -37,6 +22,7 @@ def __init__(self, config_file: str, cache_dir: str = "./struct"): self.version = self.config["version"] self.root = self.config["root"] self.Manifest = None + self.manifest_version = None if int(str(self.version).split(".")[2]) >= 7: log(f"Version: {self.version} >= 7") self.Platform = self.config.get("platform") @@ -46,11 +32,16 @@ def __init__(self, config_file: str, cache_dir: str = "./struct"): self.Branch = self.config.get("branch") self.Struct = self.config.get("struct") self.URL = f"https://raw.githubusercontent.com/{self.Owner}/{self.Repo}/refs/heads/{self.Branch}/{self.Struct}/" + if (int(str(self.version).split(".")[1]) >= 2) and (int(str(self.version).split(".")[2]) >= 14): + print("Using latest manifest attr") + self.manifest_version = self.config.get("manifest_version") + if not self.manifest_version: + raise ManifestError("Manifest version is missing in the configuration. Add tag 'manifest_version' to the config. 0.2.14+") else: self.URL = self.config.get("URL") - self.Manifest = self.config.get("manifest") - if self.Manifest: - self.Manifest = KVManifestLoader(f"{self.URL}{self.Manifest}", self.cache_dir, self.root) + manifest_file = self.config.get("manifest") + if manifest_file: + self.Manifest = KVManifestLoader(f"{self.URL}{manifest_file}", self.cache_dir, self.root, self.manifest_version or get_version()) else: raise ManifestError("Manifest file is missing in the configuration.") else: @@ -94,8 +85,8 @@ def from_namespace(self, namespace: str) -> KVProcessor: if self.Manifest: log(f"Using Manifest to load KVProcessor from namespace: {namespace}") - if namespace in self.Manifest.namespace_overides: - namespace = self.Manifest.namespace_overides[namespace] + if namespace in self.Manifest.namespace_overrides: + namespace = self.Manifest.namespace_overrides[namespace] log(f"Namespace overridden to: {namespace}") else: log(f"Namespace not found in manifest, using original: {namespace}") @@ -136,4 +127,4 @@ def list_available_namespaces(self) -> list: """List all available namespaces from the manifest.""" if not self.Manifest: raise ManifestError("Manifest is not loaded.") - return list(self.Manifest.namespace_overides.keys()) + return list(self.Manifest.namespace_overrides.keys()) diff --git a/kvprocessor/kvtypemap.py b/kvprocessor/kvtypemap.py new file mode 100644 index 0000000..2158c2d --- /dev/null +++ b/kvprocessor/kvtypemap.py @@ -0,0 +1,73 @@ +from typing import Any +import datetime +import decimal + +type_map: dict[str, Any] = { + 'string': str, + 'int': int, + 'float': float, + 'bool': bool, + 'none': type(None), + 'list': list, + 'dict': dict, + 'tuple': tuple, + 'set': set, + 'object': object, + 'any': Any, + 'str': str, + 'datetime': datetime.datetime, + 'date': datetime.date, + 'time': datetime.time, + 'decimal': decimal.Decimal, +} + +def get_type_map() -> dict[str, Any]: + """ + Returns the type map for KV types. + + :return: A dictionary mapping KV types to Python types. + """ + return type_map + +def set_type_map(new_type_map: dict[str, Any]) -> None: + """ + Replaces the current type map with a new one. + + :param new_type_map: A dictionary mapping KV types to Python types. + """ + global type_map + type_map = new_type_map + +def add_type_map(key: str, value: Any) -> None: + """ + Adds or overrides a single key-value pair in the type map. + + :param key: The key to add or override in the type map. + :param value: The Python type to associate with the key. + """ + type_map[key] = value + +def remove_type_map(key: str) -> None: + """ + Removes a key from the type map if it exists. + + :param key: The key to remove from the type map. + """ + if key in type_map: + del type_map[key] + +def has_type_map(key: str) -> bool: + """ + Checks if a key exists in the type map. + + :param key: The key to check in the type map. + :return: True if the key exists, False otherwise. + """ + return key in type_map + +def clear_type_map() -> None: + """ + Clears all entries in the type map. + """ + global type_map + type_map.clear() diff --git a/kvprocessor/kvvalidator.py b/kvprocessor/kvvalidator.py index 3deb087..c37c755 100644 --- a/kvprocessor/kvvalidator.py +++ b/kvprocessor/kvvalidator.py @@ -1,19 +1,56 @@ -from kvprocessor.errors import InvalidKVFileError import re +from kvprocessor.util.errors import InvalidKVFileError +from kvprocessor.kvtypemap import get_type_map +from kvprocessor.util.warnings import deprecated +class KVFileValidator(): + def __init__(self): + self.file_path: str = None + + def validate_kv_file(self, file_path: str) -> bool: + if self.file_path is None: + self.file_path = file_path + try: + with open(self.file_path, 'r') as file: + for i, line in enumerate(file, start=1): + line = line.strip() + if not line or line.startswith('#'): + continue + if line.split("#"): + line = line.split("#")[0].strip() + match = re.match(r'(\w+)<([\w\|]+)>:([\w+]+|none)', line) + if not match: + raise InvalidKVFileError(f"Invalid .kv file format in line {i}: {line}") + return True + except FileNotFoundError: + raise FileNotFoundError(f"KV file not found: {self.file_path}") + + def validate_kv_key(self, key: str) -> bool: + """Validate a single key in a .kv file.""" + match = re.match(r'(\w+)<([\w\|]+)>:([\w+]+|none)', key) + if not match: + raise InvalidKVFileError(f"Invalid key format: {key}") + return True + + def validate_kv_value(self, value: str, expected_types: list) -> bool: + """Validate a value against expected types.""" + type_map = get_type_map() + for type_name in expected_types: + if type_name not in type_map: + raise ValueError(f"Unsupported type: {type_name}") + if isinstance(value, type_map[type_name]): + return True + return False + +@deprecated def validate_kv_file(file_path: str) -> bool: """Validate the syntax of a .kv file.""" - try: - with open(file_path, 'r') as file: - for i, line in enumerate(file, start=1): - line = line.strip() - if not line or line.startswith('#'): - continue - if line.split("#"): - line = line.split("#")[0].strip() - match = re.match(r'(\w+)<([\w\|]+)>:([\w+]+|none)', line) - if not match: - raise InvalidKVFileError(f"Invalid .kv file format in line {i}: {line}") - return True - except FileNotFoundError: - raise FileNotFoundError(f"KV file not found: {file_path}") \ No newline at end of file + return KVFileValidator().validate_kv_file(file_path) + +@deprecated +def validate_kv_key(key: str) -> bool: + return KVFileValidator().validate_kv_key(key) + +@deprecated +def validate_kv_value(value: str, expected_types: list) -> bool: + return KVFileValidator().validate_kv_value(value, expected_types) \ No newline at end of file diff --git a/kvprocessor/kvversionmanager.py b/kvprocessor/kvversionmanager.py new file mode 100644 index 0000000..9383c9f --- /dev/null +++ b/kvprocessor/kvversionmanager.py @@ -0,0 +1,35 @@ +import os +import shutil +from datetime import datetime + +class KVVersionManager: + """Manages versions of .kv files.""" + + def __init__(self, version_dir: str = "./versions"): + self.version_dir = version_dir + os.makedirs(self.version_dir, exist_ok=True) + + def save_version(self, file_path: str): + """Save a version of the .kv file with a timestamp.""" + if not os.path.exists(file_path): + raise FileNotFoundError(f"File not found: {file_path}") + timestamp = datetime.now().strftime("%Y%m%d%H%M%S") + file_name = os.path.basename(file_path) + versioned_file = os.path.join(self.version_dir, f"{file_name}.{timestamp}") + shutil.copy(file_path, versioned_file) + return versioned_file + + def list_versions(self, file_name: str) -> list: + """List all saved versions of a .kv file.""" + versions = [] + for file in os.listdir(self.version_dir): + if file.startswith(file_name): + versions.append(file) + return sorted(versions) + + def restore_version(self, file_name: str, timestamp: str, restore_path: str): + """Restore a specific version of a .kv file.""" + versioned_file = os.path.join(self.version_dir, f"{file_name}.{timestamp}") + if not os.path.exists(versioned_file): + raise FileNotFoundError(f"Version not found: {versioned_file}") + shutil.copy(versioned_file, restore_path) \ No newline at end of file diff --git a/kvprocessor/log.py b/kvprocessor/log.py deleted file mode 100644 index 3c0d0bf..0000000 --- a/kvprocessor/log.py +++ /dev/null @@ -1,5 +0,0 @@ -import os - -def log(message: str): - if os.environ.get("DEBUG"): - print(message) \ No newline at end of file diff --git a/kvprocessor/errors.py b/kvprocessor/util/errors.py similarity index 55% rename from kvprocessor/errors.py rename to kvprocessor/util/errors.py index a27f711..9602ebd 100644 --- a/kvprocessor/errors.py +++ b/kvprocessor/util/errors.py @@ -16,4 +16,22 @@ class NamespaceNotFoundError(KVProcessorError): class InvalidNamespaceError(KVProcessorError): """Raised when a namespace is invalid or malformed.""" + pass + +# Legacy errors + +class KVStructLoaderError(Exception): + """Base exception for KVStructLoader.""" + pass + +class ConfigFetchError(KVStructLoaderError): + """Raised when there is an error fetching the configuration.""" + pass + +class KVFetchError(KVStructLoaderError): + """Raised when there is an error fetching a KV file.""" + pass + +class ManifestError(KVStructLoaderError): + """Raised when there is an issue with the manifest.""" pass \ No newline at end of file diff --git a/kvprocessor/util/filedownloader.py b/kvprocessor/util/filedownloader.py new file mode 100644 index 0000000..d534696 --- /dev/null +++ b/kvprocessor/util/filedownloader.py @@ -0,0 +1,11 @@ +import io +import os +import requests + +def download_file(url: str, destination: str): + response = requests.get(url, stream=True) + response.raise_for_status() + os.makedirs(os.path.dirname(destination), exist_ok=True) + with open(destination, 'wb') as fetched_file: + for chunk in response.iter_content(chunk_size=8192): + fetched_file.write(chunk) \ No newline at end of file diff --git a/kvprocessor/util/log.py b/kvprocessor/util/log.py new file mode 100644 index 0000000..f24647c --- /dev/null +++ b/kvprocessor/util/log.py @@ -0,0 +1,23 @@ +import logging + +# Configure logging +logging.basicConfig( + level=logging.DEBUG, + format='%(asctime)s - %(levelname)s - %(message)s', + handlers=[ + logging.StreamHandler(), + logging.FileHandler("kvprocessor.log", mode="a") + ] +) + +def log(message: str): + logging.info(message) + +def log_error(message: str): + logging.error(message) + +def log_debug(message: str): + logging.debug(message) + +def log_warning(message: str): + logging.warning(message) \ No newline at end of file diff --git a/kvprocessor/util/warnings.py b/kvprocessor/util/warnings.py new file mode 100644 index 0000000..0666e6d --- /dev/null +++ b/kvprocessor/util/warnings.py @@ -0,0 +1,24 @@ +import warnings + +def deprecated(func): + def wrapper(*args, **kwargs): + warnings.warn(f"{func.__name__} is deprecated and will be removed in a future version.", + DeprecationWarning, stacklevel=2) + return func(*args, **kwargs) + return wrapper + +def ignore_warnings(func): + """ + Decorator to ignore warnings in a function. + + Args: + func (callable): The function to decorate. + + Returns: + callable: The decorated function. + """ + def wrapper(*args, **kwargs): + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + return func(*args, **kwargs) + return wrapper \ No newline at end of file diff --git a/mypy.ini b/mypy.ini new file mode 100644 index 0000000..9032648 --- /dev/null +++ b/mypy.ini @@ -0,0 +1,21 @@ +[mypy] +# Specify the files and directories to check +files = . + +# Disallow the use of untyped function definitions +disallow_untyped_defs = False + +# Disallow untyped calls to functions +disallow_untyped_calls = False + +# Require type hints for function arguments and return values +strict = False + +# Ignore missing imports (useful if some dependencies don't have type hints) +ignore_missing_imports = True + +# Show error codes in the output +show_error_codes = True + +# Cache directory for mypy +cache_dir = .mypy_cache \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 3cd103d..62f01e1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "kvprocessor" -version = "0.1.12" +version = "0.2.14" description = "A Python package for processing and validating configuration dictionaries against a custom .kv file format" readme = "README.md" authors = [{name = "connor33341", email = "connor@connor33341.dev"}] diff --git a/test.sh b/test.sh index cbcf990..4c1c16c 100644 --- a/test.sh +++ b/test.sh @@ -1,4 +1,108 @@ +#!/bin/bash + +# CONFIGURATION +LINTING="false" +TYPE_CHECKING="true" +SECURITY_CHECKING="false" +CLEANUP="true" + +# Step 1: Build the project +echo "Starting the build process..." bash build.sh -echo "Testing" -python -m pip install -r test/requirements.txt -python test/test.py \ No newline at end of file +if [ $? -ne 0 ]; then + echo "Build failed. Exiting." + exit 1 +fi +echo "Build completed successfully." + +# Step 2: Install dependencies +echo "Installing test dependencies..." +python3 -m pip install --upgrade pip +if [ $? -ne 0 ]; then + echo "Failed to upgrade pip. Exiting." + exit 1 +fi + +python3 -m pip install -r test/requirements.txt +if [ $? -ne 0 ]; then + echo "Failed to install dependencies. Exiting." + exit 1 +fi +echo "Dependencies installed successfully." + +# Step 3: Run unit tests +echo "Running unit tests..." +python3 test/test.py +if [ $? -ne 0 ]; then + echo "Unit tests failed. Exiting." + exit 1 +fi +echo "Unit tests passed successfully." + +# Step 4: Run linting +if [ "$LINTING" = "true" ]; then + echo "Running lint checks..." + python3 -m pip install flake8 + if [ $? -ne 0 ]; then + echo "Failed to install flake8. Exiting." + exit 1 + fi + + flake8 . + if [ $? -ne 0 ]; then + echo "Linting failed. Please fix the issues." + exit 1 + fi + echo "Linting passed successfully." +fi + +# Step 5: Run type checks +if [ "$TYPE_CHECKING" = "true" ]; then + echo "Running type checks with mypy..." + python3 -m pip install mypy + if [ $? -ne 0 ]; then + echo "Failed to install mypy. Exiting." + exit 1 + fi + + mypy --install-types --non-interactive + if [ $? -ne 0 ]; then + echo "Failed to install types for mypy. Exiting." + exit 1 + fi + + mypy . + if [ $? -ne 0 ]; then + echo "Type checks failed. Please fix the issues." + exit 1 + fi + echo "Type checks passed successfully." +fi + +# Step 6: Run security checks +if [ "$SECURITY_CHECKING" = "true" ]; then + echo "Running security checks with bandit..." + python3 -m pip install bandit + if [ $? -ne 0 ]; then + echo "Failed to install bandit. Exiting." + exit 1 + fi + + bandit -r . + if [ $? -ne 0 ]; then + echo "Security checks failed. Please fix the issues." + exit 1 + fi + echo "Security checks passed successfully." +fi + +# Step 7: Clean up temporary files +if [ "$CLEANUP" = "true" ]; then + echo "Cleaning up temporary files..." + find . -type f -name "*.pyc" -delete + find . -type d -name "__pycache__" -exec rm -r {} + + echo "Cleanup completed." +fi + +# Final message +echo "All steps completed successfully." \ No newline at end of file diff --git a/test/expirament.py b/test/expirament.py index 46d6295..3558bae 100644 --- a/test/expirament.py +++ b/test/expirament.py @@ -1,9 +1,65 @@ from kvprocessor import KVProcessor, KVStructLoader import json +import sys + +def display_menu(): + print("\nMenu:") + print("1. Load Namespace") + print("2. Process Data") + print("3. Export Data") + print("4. Exit") + +def load_namespace(kv_struct_loader): + namespace = input("Enter Namespace: ") + try: + kv_processor = kv_struct_loader.from_namespace(namespace) + print(f"Namespace '{namespace}' loaded successfully.") + return kv_processor + except Exception as e: + print(f"Error loading namespace: {e}") + return None + +def process_data(kv_processor): + if not kv_processor: + print("No namespace loaded. Please load a namespace first.") + return + try: + data = input("Enter data to process (JSON format): ") + parsed_data = json.loads(data) + processed_data = kv_processor.process(parsed_data) + print("Processed Data:") + print(json.dumps(processed_data, indent=4)) + except Exception as e: + print(f"Error processing data: {e}") + +def export_data(kv_processor): + if not kv_processor: + print("No namespace loaded. Please load a namespace first.") + return + try: + export_path = input("Enter export file path: ") + kv_processor.export(export_path) + print(f"Data exported successfully to {export_path}.") + except Exception as e: + print(f"Error exporting data: {e}") if __name__ == "__main__": - print("Init") - Running = True - while Running: - kv_struct_loader = KVStructLoader("https://github.com/Voxa-Communications/VoxaCommunicaitons-Structures/raw/refs/heads/main/struct/config.json") - kv_processor: KVProcessor = kv_struct_loader.from_namespace(str(input("Namespace: "))) \ No newline at end of file + print("KV Processor Initialized") + kv_struct_loader = KVStructLoader("https://github.com/Voxa-Communications/VoxaCommunicaitons-Structures/raw/refs/heads/main/struct/config.json") + kv_processor = None + + while True: + display_menu() + choice = input("Enter your choice: ") + + if choice == "1": + kv_processor = load_namespace(kv_struct_loader) + elif choice == "2": + process_data(kv_processor) + elif choice == "3": + export_data(kv_processor) + elif choice == "4": + print("Exiting KV Processor.") + sys.exit(0) + else: + print("Invalid choice. Please try again.") \ No newline at end of file diff --git a/test/requirements.txt b/test/requirements.txt index 40448e6..4524122 100644 --- a/test/requirements.txt +++ b/test/requirements.txt @@ -1 +1,2 @@ -dotenv \ No newline at end of file +dotenv +pytest \ No newline at end of file diff --git a/test/restored_test.kv b/test/restored_test.kv new file mode 100644 index 0000000..490ec8d --- /dev/null +++ b/test/restored_test.kv @@ -0,0 +1,8 @@ +DATABASE_NAME:none +DATABASE_USER:none +DATABASE_PASSWORD:none +DATABASE_HOST:none +DATABASE_PORT:none +DATABASE_DRIVER:mysql+mysqlconnector +DATABASE_DIALECT:none +KV_FILE_PATH:config/env.kv \ No newline at end of file diff --git a/test/test.py b/test/test.py index 63a5e64..cd022f8 100644 --- a/test/test.py +++ b/test/test.py @@ -1,6 +1,15 @@ import os import dotenv +import pytest +import datetime +import decimal +import subprocess from kvprocessor import LoadEnv, KVProcessor, KVStructLoader +from kvprocessor.kvfileutils import search_kv_files, copy_kv_file, delete_kv_file +from kvprocessor.kvversionmanager import KVVersionManager +from kvprocessor.util.errors import ManifestError +from kvprocessor.kvtypemap import get_type_map +from kvprocessor.kvmanifestloader import KVManifestLoader, ManifestError dotenv.load_dotenv() # Load the .env file def test_file(): @@ -24,6 +33,179 @@ def test_struct_loader(): } validated_config = kv_processor.process_config(user_settings) # Verifies that those env varibles exist and are of the correct type print(validated_config) + +def test_file_operations_v2(): + print("Testing file operations") + kv_files = search_kv_files("test") + print("Found .kv files:", kv_files) + + if kv_files: + test_file = kv_files[0] + copy_path = "test/copy_test.kv" + copy_kv_file(test_file, copy_path) + print(f"Copied {test_file} to {copy_path}") + + delete_kv_file(copy_path) + print(f"Deleted {copy_path}") + +def test_version_manager_v2(): + print("Testing version manager") + version_manager = KVVersionManager("test/versions") + + test_file = "test/test.kv" + versioned_file = version_manager.save_version(test_file) + print(f"Saved version: {versioned_file}") + + versions = version_manager.list_versions("test.kv") + print("Available versions:", versions) + + if versions: + restore_path = "test/restored_test.kv" + version_manager.restore_version("test.kv", versions[0].split(".")[-1], restore_path) + print(f"Restored version to: {restore_path}") + +@pytest.fixture +def kv_processor(): + kv_file_path = "test/test.kv" # Directory to .kv file + return KVProcessor(kv_file_path) + +@pytest.fixture +def kv_struct_loader(): + return KVStructLoader("https://github.com/Voxa-Communications/VoxaCommunicaitons-Structures/raw/refs/heads/main/struct/config.json") + +def test_kv_processor_return_names(kv_processor): + kv_keys = kv_processor.return_names() # Gets the keys (VARIBLENAME) from the .kv file + assert isinstance(kv_keys, list) + assert len(kv_keys) > 0 + +def test_kv_processor_process_config(kv_processor): + kv_keys = kv_processor.return_names() + env_list = LoadEnv(kv_keys) # Loads all the ENV variables that match those keys + validated_config = kv_processor.process_config(env_list) # Verifies that those env variables exist and are of the correct type + assert isinstance(validated_config, dict) + +def test_kv_struct_loader_root(kv_struct_loader): + assert kv_struct_loader.root == "voxa" + +def test_kv_struct_loader_url(kv_struct_loader): + assert kv_struct_loader.URL.startswith("https://") + +def test_kv_struct_loader_namespace(kv_struct_loader): + kv_processor = kv_struct_loader.from_namespace("voxa.api.user.user_settings") + user_settings = { + "2FA_ENABLED": True, + "TELEMETRY": False, + "AGE": "25", + "LANGUAGE": "en", + } + validated_config = kv_processor.process_config(user_settings) + assert isinstance(validated_config, dict) + assert validated_config["2FA_ENABLED"] is True + assert validated_config["TELEMETRY"] is False + +def test_file_operations(): + kv_files = search_kv_files("test") + assert len(kv_files) > 0 + + test_file = kv_files[0] + copy_path = "test/copy_test.kv" + copy_kv_file(test_file, copy_path) + assert os.path.exists(copy_path) + + delete_kv_file(copy_path) + assert not os.path.exists(copy_path) + +def test_version_manager(): + version_manager = KVVersionManager("test/versions") + + test_file = "test/test.kv" + versioned_file = version_manager.save_version(test_file) + assert os.path.exists(versioned_file) + + versions = version_manager.list_versions("test.kv") + assert len(versions) > 0 + + restore_path = "test/restored_test.kv" + version_manager.restore_version("test.kv", versions[0].split(".")[-1], restore_path) + assert os.path.exists(restore_path) + delete_kv_file(restore_path) + +def test_additional_data_types(): + type_map = get_type_map() + assert 'datetime' in type_map + assert 'date' in type_map + assert 'time' in type_map + assert 'decimal' in type_map + assert type_map['datetime'] == datetime.datetime + assert type_map['decimal'] == decimal.Decimal + +def test_validate_manifest(): + valid_manifest_content = """# A valid manifest + namespace1:namespace2 + namespace3:namespace4 + """ + invalid_manifest_content = """# An invalid manifest + namespace1 namespace2 + """ + + # Write valid manifest to a temporary file + with open("test_valid_manifest.txt", "w") as file: + file.write(valid_manifest_content) + + # Write invalid manifest to a temporary file + with open("test_invalid_manifest.txt", "w") as file: + file.write(invalid_manifest_content) + + try: + loader = KVManifestLoader("test_valid_manifest.txt", root="test") + loader.validate_manifest() # Should pass without exceptions + + loader = KVManifestLoader("test_invalid_manifest.txt", root="test") + try: + loader.validate_manifest() + except ManifestError as e: + assert "Invalid manifest format" in str(e) + finally: + os.remove("test_valid_manifest.txt") + os.remove("test_invalid_manifest.txt") + +def test_list_namespaces(): + manifest_content = """# Example manifest + namespace1:namespace2 + namespace3:namespace4 + """ + + # Write manifest to a temporary file + with open("test_manifest.txt", "w") as file: + file.write(manifest_content) + + try: + loader = KVManifestLoader("test_manifest.txt", root="test") + namespaces = loader.list_namespaces() + assert namespaces == ["namespace1", "namespace3"] + finally: + os.remove("test_manifest.txt") + +def test_cli_list_namespaces(): + manifest_content = """# Example manifest + namespace1:namespace2 + namespace3:namespace4 + """ + + # Write manifest to a temporary file + with open("test_manifest.txt", "w") as file: + file.write(manifest_content) + + try: + result = subprocess.run(["python3", "-m", "kvprocessor.cli", "list-namespaces", "test_manifest.txt"], capture_output=True, text=True) + assert result.returncode == 0 + assert "namespace1" in result.stdout + assert "namespace3" in result.stdout + finally: + os.remove("test_manifest.txt") + if __name__ == "__main__": test_file() - test_struct_loader() \ No newline at end of file + test_struct_loader() + test_file_operations_v2() + test_version_manager_v2() \ No newline at end of file