Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor: Remove legacy tap and target test helpers #2207

Merged
merged 1 commit into from
Jan 31, 2024
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
34 changes: 0 additions & 34 deletions singer_sdk/testing/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,6 @@

from __future__ import annotations

import typing as t
import warnings

from .config import SuiteConfig
from .factory import get_tap_test_class, get_target_test_class
from .legacy import (
Expand All @@ -17,37 +14,6 @@
)
from .runners import SingerTestRunner, TapTestRunner, TargetTestRunner


def __getattr__(name: str) -> t.Any: # noqa: ANN401
if name == "get_standard_tap_tests":
warnings.warn(
"The function singer_sdk.testing.get_standard_tap_tests is deprecated "
"and will be removed in a future release. Use get_tap_test_class instead.",
DeprecationWarning,
stacklevel=2,
)

from .legacy import get_standard_tap_tests

return get_standard_tap_tests

if name == "get_standard_target_tests":
warnings.warn(
"The function singer_sdk.testing.get_standard_target_tests is deprecated "
"and will be removed in a future release. Use get_target_test_class "
"instead.",
DeprecationWarning,
stacklevel=2,
)

from .legacy import get_standard_target_tests

return get_standard_target_tests

msg = f"module {__name__} has no attribute {name}"
raise AttributeError(msg)


__all__ = [
"get_tap_test_class",
"get_target_test_class",
Expand Down
97 changes: 0 additions & 97 deletions singer_sdk/testing/legacy.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,103 +14,6 @@
from singer_sdk.target_base import Target


def get_standard_tap_tests( # noqa: C901
tap_class: type[Tap],
config: dict | None = None,
) -> list[t.Callable]:
"""Return callable pytest which executes simple discovery and connection tests.

Args:
tap_class: TODO
config: TODO

Returns:
TODO
"""

def _test_cli_prints() -> None:
# Initialize with basic config
tap1: Tap = tap_class(config=config, parse_env_config=True)
# Test CLI prints
tap1.print_version()
tap1.print_about()
tap1.print_about(output_format="json")

def _test_discovery() -> None:
catalog1 = _get_tap_catalog(tap_class, config or {})
# Reset and re-initialize with an input catalog
tap2: Tap = tap_class(config=config, parse_env_config=True, catalog=catalog1)
assert tap2

def _test_stream_connections() -> None:
# Initialize with basic config
tap1: Tap = tap_class(config=config, parse_env_config=True)
tap1.run_connection_test()

def _test_pkeys_in_schema() -> None:
"""Verify that primary keys are actually in the stream's schema."""
tap = tap_class(config=config, parse_env_config=True)
for name, stream in tap.streams.items():
pkeys = stream.primary_keys or []
schema_props = set(stream.schema["properties"].keys())
for pkey in pkeys:
error_message = (
f"Coding error in stream '{name}': "
f"primary_key '{pkey}' is missing in schema"
)
assert pkey in schema_props, error_message

def _test_state_partitioning_keys_in_schema() -> None:
"""Verify that state partitioning keys are actually in the stream's schema."""
tap = tap_class(config=config, parse_env_config=True)
for name, stream in tap.streams.items():
sp_keys = stream.state_partitioning_keys or []
schema_props = set(stream.schema["properties"].keys())
for sp_key in sp_keys:
assert sp_key in schema_props, (
f"Coding error in stream '{name}': state_partitioning_key "
f"'{sp_key}' is missing in schema"
)

def _test_replication_keys_in_schema() -> None:
"""Verify that the replication key is actually in the stream's schema."""
tap = tap_class(config=config, parse_env_config=True)
for name, stream in tap.streams.items():
rep_key = stream.replication_key
if rep_key is None:
continue
schema_props = set(stream.schema["properties"].keys())
assert rep_key in schema_props, (
f"Coding error in stream '{name}': replication_key "
f"'{rep_key}' is missing in schema"
)

return [
_test_cli_prints,
_test_discovery,
_test_stream_connections,
_test_pkeys_in_schema,
_test_state_partitioning_keys_in_schema,
_test_replication_keys_in_schema,
]


def get_standard_target_tests(
target_class: type[Target], # noqa: ARG001
config: dict | None = None, # noqa: ARG001
) -> list[t.Callable]:
"""Return callable pytest which executes simple discovery and connection tests.

Args:
target_class: The target class to test.
config: A config dictionary for the tests.

Returns:
A list of callable tests.
"""
return []


def tap_sync_test(tap: Tap) -> tuple[io.StringIO, io.StringIO]:
"""Invokes a Tap object and return STDOUT and STDERR results in StringIO buffers.

Expand Down
18 changes: 0 additions & 18 deletions tests/core/test_testing.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,27 +2,9 @@

from __future__ import annotations

import pytest

from singer_sdk.testing.factory import BaseTestClass


def test_module_deprecations():
with pytest.deprecated_call():
from singer_sdk.testing import get_standard_tap_tests # noqa: F401

with pytest.deprecated_call():
from singer_sdk.testing import get_standard_target_tests # noqa: F401

from singer_sdk import testing

with pytest.raises(
AttributeError,
match="module singer_sdk.testing has no attribute",
):
testing.foo # noqa: B018


def test_test_class_mro():
class PluginTestClass(BaseTestClass):
pass
Expand Down
Binary file added tests/fixtures/tap.sqlite
Binary file not shown.
55 changes: 44 additions & 11 deletions tests/samples/test_tap_sqlite.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from singer_sdk import SQLStream
from singer_sdk._singerlib import MetadataMapping, StreamMetadata
from singer_sdk.testing import (
get_standard_tap_tests,
get_tap_test_class,
tap_sync_test,
tap_to_target_sync_test,
)
Expand Down Expand Up @@ -104,16 +104,6 @@ def test_sqlite_input_catalog(sqlite_sample_tap: SQLTap):
assert stream.fully_qualified_name == "main.t1"


def test_sqlite_tap_standard_tests(sqlite_sample_tap: SQLTap):
"""Run standard tap tests against Countries tap."""
tests = get_standard_tap_tests(
type(sqlite_sample_tap),
dict(sqlite_sample_tap.config),
)
for test in tests:
test()


def test_sync_sqlite_to_csv(sqlite_sample_tap: SQLTap, tmp_path: Path):
_discover_and_select_all(sqlite_sample_tap)
orig_stdout, _, _, _ = tap_to_target_sync_test(
Expand Down Expand Up @@ -144,3 +134,46 @@ def test_sqlite_state(sqlite_sample_tap_state_messages):
for message in sqlite_sample_tap_state_messages
for bookmark in message["value"]["bookmarks"].values()
)


TestSQLiteTap = get_tap_test_class(
SQLiteTap,
config={
"path_to_db": "tests/fixtures/tap.sqlite",
},
catalog={
"streams": [
{
"tap_stream_id": "main-test",
"stream": "main-test",
"table_name": "test",
"schema": {
"properties": {
"id": {"type": ["integer", "null"]},
"name": {"type": ["string", "null"]},
},
},
"metadata": [
{
"breadcrumb": [],
"metadata": {
"selected": True,
},
},
{
"breadcrumb": ["properties", "id"],
"metadata": {
"selected": True,
},
},
{
"breadcrumb": ["properties", "name"],
"metadata": {
"selected": True,
},
},
],
},
],
},
)
Loading