test_utilities.py

import datetime
import os
from pathlib import Path
from unittest import mock

import pytest

from parsons import Table
from parsons.utilities import check_env, files, json_format, sql_helpers
from parsons.utilities.datetime import date_to_timestamp, parse_date


@pytest.mark.parametrize(
    ("date", "exp_ts"),
    [
        pytest.param("2018-12-13", 1544659200),
        pytest.param("2018-12-13T00:00:00-08:00", 1544688000),
        pytest.param("", None),
        pytest.param(
            "2018-12-13 PST", None, marks=[pytest.mark.xfail(raises=ValueError, strict=True)]
        ),
    ],
)
def test_date_to_timestamp(date, exp_ts):
    assert date_to_timestamp(date) == exp_ts


def test_parse_date():
    # Test parsing an ISO8601 string
    expected = datetime.datetime(year=2020, month=1, day=1, tzinfo=datetime.timezone.utc)
    parsed = parse_date("2020-01-01T00:00:00.000 UTC")
    assert parsed == expected, parsed

    # Test parsing a unix timestamp
    parsed = parse_date(1577836800)
    assert parsed == expected, parsed

    # Test "parsing" a datetime object
    parsed = parse_date(expected)
    assert parsed == expected, parsed


#
# File utility tests (pytest-style)
#


def test_create_temp_file_for_path():
    temp_path = files.create_temp_file_for_path("some/file.gz")
    assert temp_path[-3:] == ".gz"


def test_create_temp_directory():
    temp_directory = files.create_temp_directory()
    test_file1 = Path(temp_directory) / "test.txt"
    test_file2 = Path(temp_directory) / "test2.txt"
    test_file1.write_text("TEST")
    test_file2.write_text("TEST")

    assert files.has_data(test_file1)
    assert files.has_data(test_file2)

    files.cleanup_temp_directory(temp_directory)

    # Verify the temp file no longer exists
    with pytest.raises(FileNotFoundError):
        Path(test_file1).read_bytes()


def test_close_temp_file():
    temp = files.create_temp_file()
    files.close_temp_file(temp)

    # Verify the temp file no longer exists
    with pytest.raises(FileNotFoundError):
        Path(temp).read_bytes()


def test_is_gzip_path():
    assert files.is_gzip_path("some/file.gz")
    assert not files.is_gzip_path("some/file")
    assert not files.is_gzip_path("some/file.csv")


def test_suffix_for_compression_type():
    assert files.suffix_for_compression_type(None) == ""
    assert files.suffix_for_compression_type("") == ""
    assert files.suffix_for_compression_type("gzip") == ".gz"


def test_compression_type_for_path():
    assert files.compression_type_for_path("some/file") is None
    assert files.compression_type_for_path("some/file.csv") is None
    assert files.compression_type_for_path("some/file.csv.gz") == "gzip"


def test_empty_file(tmp_path: Path):
    # Create fake files.
    empty_csv = tmp_path / "empty.csv"
    empty_csv.touch()

    full_csv = tmp_path / "full.csv"

    Table([["1"], ["a"]]).to_csv(str(full_csv))

    assert not files.has_data(empty_csv)
    assert files.has_data(full_csv)


def test_json_format():
    assert json_format.arg_format("my_arg") == "myArg"


def test_remove_empty_keys():
    # Assert key removed when None
    test_dict = {"a": None, "b": 2}
    assert json_format.remove_empty_keys(test_dict) == {"b": 2}

    # Assert key not removed when None
    test_dict = {"a": 1, "b": 2}
    assert json_format.remove_empty_keys(test_dict) == {"a": 1, "b": 2}

    # Assert that a nested empty string is removed
    test_dict = {"a": "", "b": 2}
    assert json_format.remove_empty_keys(test_dict) == {"b": 2}


def test_redact_credentials():
    # Test with quotes, escape characters, and line breaks
    test_str = r"""COPY schema.tablename
    FROM 's3://bucket/path/to/file.csv'
    credentials  'aws_access_key_id=string-\'escaped-quote;
    aws_secret_access_key='string-escape-char\\'
    MANIFEST"""

    test_result = """COPY schema.tablename
    FROM 's3://bucket/path/to/file.csv'
    CREDENTIALS REDACTED
    MANIFEST"""

    assert sql_helpers.redact_credentials(test_str) == test_result


class TestCheckEnv:
    @pytest.mark.parametrize(
        ("environment_value", "input_value", "expected_result"),
        [
            ({}, "param", "param"),
            ({"PARAM": "env_param"}, None, "env_param"),
            ({"PARAM": "env_param"}, "param", "param"),
        ],
        ids=["test_environment_field", "test_environment_env", "test_environment_field_env"],
    )
    def test_check_env_success(
        self, environment_value: dict, input_value: str | None, expected_result: str
    ):
        """Tests successful retrieval of parameters from field or environment."""
        with mock.patch.dict(os.environ, environment_value):
            result = check_env.check("PARAM", input_value)
        assert result == expected_result

    def test_check_env_returns_none(self):
        """Tests returning None when environment and value are empty and optional value is passed."""
        with mock.patch.dict(os.environ, {}, clear=True):
            result = check_env.check("PARAM", None, optional=True)
        assert result is None

    def test_environment_error(self):
        """Test check env raises error when both are missing."""
        # We ensure the environment is empty for this key to trigger the KeyError
        with (
            mock.patch.dict(os.environ, {}, clear=True),
            pytest.raises(
                KeyError,
                match="No 'PARAM' found. Store as environment variable or pass as an argument.",
            ),
        ):
            check_env.check("PARAM", None)