Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions tests/test_backend_dynamo.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,26 +5,25 @@
from __future__ import annotations

import json
import os
import time
from unittest import mock

import boto3
import pytest
from moto import mock_aws

from zebra_day.backends.dynamo import DynamoBackend, _BACKUP_DEBOUNCE_SECONDS
from zebra_day.backends.dynamo import DynamoBackend
from zebra_day.exceptions import (
ConfigError,
LabelTemplateNotFoundError,
VersionConflictError,
)


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def aws_env(monkeypatch):
"""Set fake AWS credentials for moto."""
Expand Down Expand Up @@ -60,6 +59,7 @@ def dynamo_backend(aws_env):
# Config CRUD
# ---------------------------------------------------------------------------


class TestLoadConfig:
def test_empty_table_returns_default(self, dynamo_backend):
config = dynamo_backend.load_config()
Expand Down Expand Up @@ -119,6 +119,7 @@ def test_config_exists(self, dynamo_backend):
# Template CRUD
# ---------------------------------------------------------------------------


class TestTemplateCRUD:
def test_save_and_get(self, dynamo_backend):
dynamo_backend.save_template("my_label", "^XA^FO10,10^A0N,30,30^FDHello^FS^XZ")
Expand Down Expand Up @@ -163,11 +164,11 @@ def test_template_version_conflict(self, dynamo_backend):
dynamo_backend.save_template("tpl", "v2")



# ---------------------------------------------------------------------------
# S3 Backup & Restore
# ---------------------------------------------------------------------------


class TestS3Backup:
def test_backup_creates_manifest(self, dynamo_backend):
dynamo_backend.save_config({"schema_version": "2.1.0", "labs": {"a": {}}})
Expand Down Expand Up @@ -230,6 +231,7 @@ def test_restore_round_trip(self, dynamo_backend):
# Resource Tagging
# ---------------------------------------------------------------------------


class TestResourceTagging:
def test_dynamodb_table_tags(self, aws_env):
with mock_aws():
Expand All @@ -243,7 +245,9 @@ def test_dynamodb_table_tags(self, aws_env):
backend.create_table()

# Check tags via describe
arn = backend._ddb_client.describe_table(TableName="tag-test-table")["Table"]["TableArn"]
arn = backend._ddb_client.describe_table(TableName="tag-test-table")["Table"][
"TableArn"
]
tags_resp = backend._ddb_client.list_tags_of_resource(ResourceArn=arn)
tags = {t["Key"]: t["Value"] for t in tags_resp.get("Tags", [])}
assert tags["lsmc-cost-center"] == "my-cc"
Expand Down Expand Up @@ -293,6 +297,7 @@ def test_tag_from_env(self, aws_env, monkeypatch):
# Factory & Profile Rules
# ---------------------------------------------------------------------------


class TestFromEnv:
def test_from_env_basic(self, aws_env, monkeypatch):
monkeypatch.setenv("ZEBRA_DAY_DYNAMO_TABLE", "my-table")
Expand Down Expand Up @@ -341,6 +346,7 @@ def test_no_explicit_default_profile(self, aws_env):
# Status
# ---------------------------------------------------------------------------


class TestStatus:
def test_get_status(self, dynamo_backend):
dynamo_backend.save_config({"schema_version": "2.1.0", "labs": {}})
Expand All @@ -355,6 +361,7 @@ def test_get_status(self, dynamo_backend):
# Integration: zpl() with DynamoBackend
# ---------------------------------------------------------------------------


class TestZplIntegration:
def test_zpl_init_with_backend(self, dynamo_backend):
from zebra_day.print_mgr import zpl
Expand Down Expand Up @@ -419,6 +426,7 @@ def test_env_var_selection(self, aws_env, monkeypatch):
# AWS Permission Checks
# ---------------------------------------------------------------------------


class TestCheckAWSPermissions:
def test_all_checks_pass_with_resources(self, dynamo_backend):
result = dynamo_backend.check_aws_permissions()
Expand Down
141 changes: 84 additions & 57 deletions tests/test_cli_dynamo.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,7 @@
from __future__ import annotations

import json
import os
from pathlib import Path

import boto3
import pytest
from moto import mock_aws
from typer.testing import CliRunner
Expand Down Expand Up @@ -67,10 +64,14 @@ def test_init_success(self, aws_env):
result = runner.invoke(
app,
[
"dynamo", "init",
"--table-name", "test-zebra-config",
"--s3-bucket", "test-backup-bucket",
"--region", "us-east-1",
"dynamo",
"init",
"--table-name",
"test-zebra-config",
"--s3-bucket",
"test-backup-bucket",
"--region",
"us-east-1",
],
)
assert result.exit_code == 0, result.output
Expand All @@ -94,10 +95,14 @@ def test_init_with_s3_config_file(self, aws_env, tmp_path):
result = runner.invoke(
app,
[
"dynamo", "init",
"--table-name", "test-zebra-config",
"--region", "us-east-1",
"--s3-config-file", str(cfg),
"dynamo",
"init",
"--table-name",
"test-zebra-config",
"--region",
"us-east-1",
"--s3-config-file",
str(cfg),
],
)
assert result.exit_code == 0, result.output
Expand All @@ -110,11 +115,16 @@ def test_init_s3_bucket_flag_overrides_config_file(self, aws_env, tmp_path):
result = runner.invoke(
app,
[
"dynamo", "init",
"--table-name", "test-zebra-config",
"--region", "us-east-1",
"--s3-bucket", "flag-bucket",
"--s3-config-file", str(cfg),
"dynamo",
"init",
"--table-name",
"test-zebra-config",
"--region",
"us-east-1",
"--s3-bucket",
"flag-bucket",
"--s3-config-file",
str(cfg),
],
)
assert result.exit_code == 0, result.output
Expand All @@ -127,10 +137,14 @@ def test_init_bad_s3_config_file(self, aws_env, tmp_path):
result = runner.invoke(
app,
[
"dynamo", "init",
"--table-name", "test-zebra-config",
"--region", "us-east-1",
"--s3-config-file", str(cfg),
"dynamo",
"init",
"--table-name",
"test-zebra-config",
"--region",
"us-east-1",
"--s3-config-file",
str(cfg),
],
)
assert result.exit_code == 1
Expand All @@ -140,10 +154,14 @@ def test_init_nonexistent_s3_config_file(self, aws_env):
result = runner.invoke(
app,
[
"dynamo", "init",
"--table-name", "test-zebra-config",
"--region", "us-east-1",
"--s3-config-file", "/tmp/no-such-file-12345.json",
"dynamo",
"init",
"--table-name",
"test-zebra-config",
"--region",
"us-east-1",
"--s3-config-file",
"/tmp/no-such-file-12345.json",
],
)
assert result.exit_code == 1
Expand All @@ -154,24 +172,35 @@ def test_init_shows_permission_checks(self, aws_env):
result = runner.invoke(
app,
[
"dynamo", "init",
"--table-name", "test-zebra-config",
"--s3-bucket", "test-backup-bucket",
"--region", "us-east-1",
"dynamo",
"init",
"--table-name",
"test-zebra-config",
"--s3-bucket",
"test-backup-bucket",
"--region",
"us-east-1",
],
)
assert result.exit_code == 0, result.output
assert "permission checks passed" in result.output.lower() or "Checking AWS" in result.output
assert (
"permission checks passed" in result.output.lower()
or "Checking AWS" in result.output
)

def test_init_skip_checks(self, aws_env):
with mock_aws():
result = runner.invoke(
app,
[
"dynamo", "init",
"--table-name", "test-zebra-config",
"--s3-bucket", "test-backup-bucket",
"--region", "us-east-1",
"dynamo",
"init",
"--table-name",
"test-zebra-config",
"--s3-bucket",
"test-backup-bucket",
"--region",
"us-east-1",
"--skip-checks",
],
)
Expand Down Expand Up @@ -225,9 +254,12 @@ def test_bootstrap_with_config_and_templates(self, aws_env, tmp_path):
result = runner.invoke(
app,
[
"dynamo", "bootstrap",
"--config-file", str(cfg),
"--templates-dir", str(tpl_dir),
"dynamo",
"bootstrap",
"--config-file",
str(cfg),
"--templates-dir",
str(tpl_dir),
"--no-include-package",
],
)
Expand All @@ -237,8 +269,6 @@ def test_bootstrap_with_config_and_templates(self, aws_env, tmp_path):
assert "Backup written" in result.output




# ---------------------------------------------------------------------------
# export
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -381,7 +411,6 @@ def test_destroy_success(self, aws_env):
assert "Backups preserved" in result.output



# ---------------------------------------------------------------------------
# Interactive S3 bucket prompt
# ---------------------------------------------------------------------------
Expand All @@ -398,9 +427,7 @@ def test_status_prompts_for_bucket_interactively(self, aws_env, monkeypatch):
with mock_aws():
_provision_resources()
# Provide bucket name via CliRunner input
result = runner.invoke(
app, ["dynamo", "status"], input="test-backup-bucket\n"
)
result = runner.invoke(app, ["dynamo", "status"], input="test-backup-bucket\n")
assert result.exit_code == 0, result.output
assert "test-zebra-config" in result.output

Expand Down Expand Up @@ -429,9 +456,7 @@ def test_backup_prompts_for_bucket(self, aws_env, monkeypatch):
with mock_aws():
backend = _provision_resources()
backend.save_config({"labs": {}})
result = runner.invoke(
app, ["dynamo", "backup"], input="test-backup-bucket\n"
)
result = runner.invoke(app, ["dynamo", "backup"], input="test-backup-bucket\n")
assert result.exit_code == 0, result.output
assert "Backup written" in result.output

Expand All @@ -443,9 +468,12 @@ def test_init_prompts_for_bucket(self, aws_env, monkeypatch):
result = runner.invoke(
app,
[
"dynamo", "init",
"--table-name", "test-zebra-config",
"--region", "us-east-1",
"dynamo",
"init",
"--table-name",
"test-zebra-config",
"--region",
"us-east-1",
],
input="test-backup-bucket\n",
)
Expand All @@ -467,6 +495,7 @@ def test_status_create_s3_if_missing(self, aws_env, monkeypatch):
_provision_resources()
# Delete the bucket so it doesn't exist
import boto3

s3 = boto3.client("s3", region_name="us-east-1")
# Empty and delete the bucket
try:
Expand All @@ -479,9 +508,7 @@ def test_status_create_s3_if_missing(self, aws_env, monkeypatch):

# Use a new bucket name via env
monkeypatch.setenv("ZEBRA_DAY_S3_BACKUP_BUCKET", "new-auto-bucket")
result = runner.invoke(
app, ["dynamo", "status", "--create-s3-if-missing"]
)
result = runner.invoke(app, ["dynamo", "status", "--create-s3-if-missing"])
assert result.exit_code == 0, result.output
# Bucket should have been created
assert "created" in result.output.lower() or "test-zebra-config" in result.output
Expand All @@ -494,9 +521,7 @@ def test_backup_create_s3_if_missing(self, aws_env, monkeypatch):

# Point to a new bucket that doesn't exist
monkeypatch.setenv("ZEBRA_DAY_S3_BACKUP_BUCKET", "auto-backup-bucket")
result = runner.invoke(
app, ["dynamo", "backup", "--create-s3-if-missing"]
)
result = runner.invoke(app, ["dynamo", "backup", "--create-s3-if-missing"])
assert result.exit_code == 0, result.output
assert "Backup written" in result.output

Expand All @@ -510,11 +535,13 @@ def test_bootstrap_create_s3_if_missing(self, aws_env, monkeypatch, tmp_path):
result = runner.invoke(
app,
[
"dynamo", "bootstrap",
"--config-file", str(cfg),
"dynamo",
"bootstrap",
"--config-file",
str(cfg),
"--no-include-package",
"--create-s3-if-missing",
],
)
assert result.exit_code == 0, result.output
assert "created" in result.output.lower() or "Config uploaded" in result.output
assert "created" in result.output.lower() or "Config uploaded" in result.output
6 changes: 4 additions & 2 deletions tests/test_core_functions.py
Original file line number Diff line number Diff line change
Expand Up @@ -354,8 +354,10 @@ def test_probe_accepts_valid_stub(self):
zp = zdpm.zpl()
# Mock HTTPConnection and HTTPSConnection so the 255-IP loop
# completes instantly without real network calls.
with mock.patch("http.client.HTTPConnection") as mock_http, \
mock.patch("http.client.HTTPSConnection") as mock_https:
with (
mock.patch("http.client.HTTPConnection") as mock_http,
mock.patch("http.client.HTTPSConnection") as mock_https,
):
# Make every connection attempt raise immediately (no printer)
mock_http.return_value.request.side_effect = OSError("mocked")
mock_https.return_value.request.side_effect = OSError("mocked")
Expand Down
Loading
Loading