From 8cb0dac58ea34be834f44ae13878761e4a71418d Mon Sep 17 00:00:00 2001 From: Ivan Ogasawara Date: Thu, 21 Dec 2023 21:21:25 -0400 Subject: [PATCH] generate lock data individually for each profile --- src/envers/core.py | 63 +++++++++++++++++++++++++++-------------- src/envers/crypt.py | 10 +++---- tests/test_core.py | 68 +++++++++++++++++++++++++++++++++------------ 3 files changed, 98 insertions(+), 43 deletions(-) diff --git a/src/envers/core.py b/src/envers/core.py index 90cc6d5..6254ea9 100644 --- a/src/envers/core.py +++ b/src/envers/core.py @@ -6,7 +6,7 @@ import os from pathlib import Path -from typing import Any +from typing import Any, Optional import typer import yaml # type: ignore @@ -52,7 +52,7 @@ def merge_dicts( # constants ENVERS_SPEC_FILENAME = "specs.yaml" -ENVERS_DATA_FILENAME = "data.lock" +# ENVERS_DATA_FILENAME = "data.lock" def escape_template_tag(v: str) -> str: @@ -68,8 +68,10 @@ def unescape_template_tag(v: str) -> str: class Envers: """EnversBase defined the base structure for the Envers classes.""" - def _read_data_file(self, password: str = "") -> dict[str, Any]: - data_file = Path(".envers") / ENVERS_DATA_FILENAME + def _read_data_file( + self, profile: str, password: str = "" + ) -> dict[str, Any]: + data_file = Path(".envers") / "data" / f"{profile}.lock" with open(data_file, "r") as file: try: @@ -90,9 +92,11 @@ def _read_data_file(self, password: str = "") -> dict[str, Any]: return data_lock def _write_data_file( - self, data: dict[str, Any], password: str = "" + self, profile: str, data: dict[str, Any], password: str = "" ) -> None: - data_file = Path(".envers") / ENVERS_DATA_FILENAME + data_file = Path(".envers") / "data" / f"{profile}.lock" + + os.makedirs(data_file.parent, exist_ok=True) with open(data_file, "w") as file: data_content = yaml.dump(data, sort_keys=False) @@ -213,23 +217,30 @@ def draft( with open(spec_file, "w") as file: yaml.dump(specs, file, sort_keys=False) - def deploy(self, profile: str, spec: str) -> None: + def deploy( + self, profile: str, spec: str, password: Optional[str] = None + ) -> None: """ Deploy a specific version, updating the .envers/data.lock file. Parameters ---------- + profile : str + The profile to be deployed. spec : str The version number to be deployed. + password : Optional[str] + The password to be used for that profile. Returns ------- None """ specs_file = Path(".envers") / ENVERS_SPEC_FILENAME - data_file = Path(".envers") / ENVERS_DATA_FILENAME + data_file = Path(".envers") / "data" / f"{profile}.lock" - password = crypt.get_password() + if password is None: + password = crypt.get_password() if not specs_file.exists(): typer.echo("Spec file not found. Please initialize envers first.") @@ -248,7 +259,7 @@ def deploy(self, profile: str, spec: str) -> None: del spec_data["status"] if data_file.exists(): - data_lock = self._read_data_file(password) + data_lock = self._read_data_file(profile, password) if not data_lock: typer.echo("data.lock is not valid. Creating a new file.") @@ -279,13 +290,15 @@ def deploy(self, profile: str, spec: str) -> None: profile_data["files"][file_path] = file_data data_lock["releases"][spec]["data"][profile_name] = profile_data - self._write_data_file(data_lock, password) + self._write_data_file(profile, data_lock, password) with open(specs_file, "w") as file: specs["releases"][spec]["status"] = "deployed" yaml.dump(specs, file, sort_keys=False) - def profile_set(self, profile: str, spec: str) -> None: + def profile_set( + self, profile: str, spec: str, password: Optional[str] = None + ) -> None: """ Set the profile values for a given spec version. @@ -295,12 +308,14 @@ def profile_set(self, profile: str, spec: str) -> None: The name of the profile to set values for. spec : str The version of the spec to use. + password : Optional[str] + The password to be used for that profile. Returns ------- None """ - data_file = Path(".envers") / ENVERS_DATA_FILENAME + data_file = Path(".envers") / "data" / f"{profile}.lock" if not data_file.exists(): typer.echo( @@ -308,9 +323,10 @@ def profile_set(self, profile: str, spec: str) -> None: ) raise typer.Exit() - password = crypt.get_password() + if password is None: + password = crypt.get_password() - data_lock = self._read_data_file(password) + data_lock = self._read_data_file(profile, password) if not data_lock.get("releases", {}).get(spec, ""): typer.echo(f"Version {spec} not found in data.lock.") @@ -349,9 +365,11 @@ def profile_set(self, profile: str, spec: str) -> None: # Update data.lock file data_lock["releases"][spec]["data"][profile] = profile_data - self._write_data_file(data_lock, password) + self._write_data_file(profile, data_lock, password) - def profile_load(self, profile: str, spec: str) -> None: + def profile_load( + self, profile: str, spec: str, password: Optional[str] = None + ) -> None: """ Load a specific environment profile to files. @@ -364,22 +382,25 @@ def profile_load(self, profile: str, spec: str) -> None: The name of the profile to load. spec : str The version of the spec to use. + password : Optional[str] + The password to be used for that profile. Returns ------- None """ - data_lock_file = Path(".envers") / "data.lock" + data_file = Path(".envers") / "data" / f"{profile}.lock" - if not data_lock_file.exists(): + if not data_file.exists(): typer.echo( "Data lock file not found. Please deploy a version first." ) raise typer.Exit() - password = crypt.get_password() + if password is None: + password = crypt.get_password() - data_lock = self._read_data_file(password) + data_lock = self._read_data_file(profile, password) if not data_lock.get("releases", {}).get(spec, ""): typer.echo(f"Version {spec} not found in data.lock.") diff --git a/src/envers/crypt.py b/src/envers/crypt.py index 9e54923..6e4faf0 100644 --- a/src/envers/crypt.py +++ b/src/envers/crypt.py @@ -5,7 +5,7 @@ import os import sys -from typing import cast +from typing import Optional, cast import typer @@ -48,9 +48,9 @@ def generate_salt() -> bytes: return os.urandom(SALT_LENGTH) -def encrypt_data(data: str, password: str = "") -> str: +def encrypt_data(data: str, password: Optional[str] = None) -> str: """Encrypt the given data.""" - if not password: + if password is None: password = get_password() salt = generate_salt() @@ -62,9 +62,9 @@ def encrypt_data(data: str, password: str = "") -> str: return salt_hex + encrypted -def decrypt_data(data: str, password: str = "") -> str: +def decrypt_data(data: str, password: Optional[str] = None) -> str: """Decrypt the given data.""" - if not password: + if password is None: password = get_password() HEX_SALT_LENGTH = SALT_LENGTH * 2 diff --git a/tests/test_core.py b/tests/test_core.py index 4cd9b8f..fb60882 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -45,6 +45,9 @@ def setup(self): self.original_cwd = os.getcwd() os.chdir(self.temp_dir.name) + self.envers = Envers() + self.envers.init(Path(".")) + def teardown(self): """Clean up temporary data.""" os.chdir(self.original_cwd) @@ -56,10 +59,8 @@ def test_temp_dir(self): def test_draft(self): """Test draft method.""" - envers = Envers() - envers.init(Path(".")) spec_version = "1.0" - envers.draft(spec_version) + self.envers.draft(spec_version) expected_data = { "version": "0.1", @@ -80,9 +81,6 @@ def test_draft(self): def test_draft_from_spec(self, spec_v1): """Test draft method with from_spec.""" - envers = Envers() - envers.init(Path(".")) - v1 = "1.0" v2 = "2.0" @@ -91,7 +89,7 @@ def test_draft_from_spec(self, spec_v1): with open(".envers/specs.yaml", "w") as f: yaml.safe_dump(initial_specs, f) - envers.draft(v2, from_spec=v1) + self.envers.draft(v2, from_spec=v1) expected_data = copy.deepcopy(initial_specs) expected_data["releases"][v2] = copy.deepcopy(spec_v1) @@ -103,15 +101,12 @@ def test_draft_from_spec(self, spec_v1): def test_draft_from_env(self, spec_v1): """Test draft method with from_env.""" - envers = Envers() - envers.init(Path(".")) - v1 = "1.0" with open(".env", "w") as f: f.write("var=hello") - envers.draft(v1, from_env=".env") + self.envers.draft(v1, from_env=".env") expected_data = {"version": "0.1", "releases": {v1: spec_v1}} @@ -122,9 +117,6 @@ def test_draft_from_env(self, spec_v1): def test_draft_from_multi_env(self, spec_v1): """Test draft method with multiple from_env.""" - envers = Envers() - envers.init(Path(".")) - v1 = "1.0" env_path_0 = ".env" @@ -145,7 +137,7 @@ def test_draft_from_multi_env(self, spec_v1): # test with .env - envers.draft(v1, from_env=env_path_0) + self.envers.draft(v1, from_env=env_path_0) expected_data = {"version": "0.1", "releases": {v1: spec_v1}} @@ -156,7 +148,7 @@ def test_draft_from_multi_env(self, spec_v1): # test with .env, and folder1/.env - envers.draft(v1, from_env=env_path_1) + self.envers.draft(v1, from_env=env_path_1) with open(".envers/specs.yaml", "r") as f: result_data = yaml.safe_load(f) @@ -178,7 +170,7 @@ def test_draft_from_multi_env(self, spec_v1): # test with .env, and folder1/.env, and folder2/.env - envers.draft(v1, from_env=env_path_2) + self.envers.draft(v1, from_env=env_path_2) with open(".envers/specs.yaml", "r") as f: result_data = yaml.safe_load(f) @@ -197,3 +189,45 @@ def test_draft_from_multi_env(self, spec_v1): } assert expected_data == result_data + + def test_deploy(self): + """Test draft method.""" + spec_version = "1.0" + password = "Envers everywhere!" + profile = "base" + + self.envers.draft(spec_version) + self.envers.deploy( + profile=profile, spec=spec_version, password=password + ) + + data_dir = Path(self.temp_dir.name) / ".envers" / "data" + + assert data_dir.exists() + assert (data_dir / f"{profile}.lock").exists() + + def test_profile_load(self, spec_v1): + """Test draft method.""" + spec_version = "1.0" + password = "Envers everywhere!" + profile = "base" + + initial_specs = {"version": "0.1", "releases": {spec_version: spec_v1}} + + with open(".envers/specs.yaml", "w") as f: + yaml.safe_dump(initial_specs, f) + + self.envers.deploy( + profile=profile, spec=spec_version, password=password + ) + self.envers.profile_load( + profile=profile, spec=spec_version, password=password + ) + + dotenv_path = Path(self.temp_dir.name) / ".env" + assert dotenv_path.exists() + + with open(dotenv_path, "r") as f: + dotenv_content = f.read() + + assert dotenv_content == "var=hello\n"