Skip to content
New issue

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

#

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

Already on GitHub? # to your account

refactor: Break up TapTestClassFactory._annotate_test_class into simpler methods #2056

Merged
merged 1 commit into from
Nov 22, 2023
Merged

refactor: Break up TapTestClassFactory._annotate_test_class into simpler methods #2056

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
160 changes: 93 additions & 67 deletions singer_sdk/testing/factory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,20 @@
from .config import SuiteConfig
from .runners import TapTestRunner, TargetTestRunner
from .suites import (
TestSuite,
tap_stream_attribute_tests,
tap_stream_tests,
tap_tests,
target_tests,
)

if t.TYPE_CHECKING:
from singer_sdk import Tap, Target
from singer_sdk import Stream, Tap, Target
from singer_sdk.testing.templates import (
AttributeTestTemplate,
StreamTestTemplate,
TapTestTemplate,
)


class BaseTestClass:
Expand Down Expand Up @@ -144,8 +150,7 @@ def runner(self) -> TapTestRunner | TargetTestRunner:

return TapTestClass

# TODO: Refactor this. It's too long and nested.
def _annotate_test_class( # noqa: C901
def _annotate_test_class(
self,
empty_test_class: type[BaseTestClass],
test_suites: list,
Expand All @@ -163,80 +168,101 @@ def _annotate_test_class( # noqa: C901
"""
for suite in test_suites:
if suite.kind == "tap":
for test_class in suite.tests:
test = test_class()
test_name = f"test_{suite.kind}_{test.name}"
setattr(empty_test_class, test_name, test.run)
self._with_tap_tests(empty_test_class, suite)

if suite.kind in {"tap_stream", "tap_stream_attribute"}:
streams = list(test_runner.new_tap().streams.values())

if suite.kind == "tap_stream":
params = [
self._with_stream_tests(empty_test_class, suite, streams)

if suite.kind == "tap_stream_attribute":
self._with_stream_attribute_tests(empty_test_class, suite, streams)

return empty_test_class

def _with_tap_tests(
self,
empty_test_class: type[BaseTestClass],
suite: TestSuite[TapTestTemplate],
) -> None:
for test_class in suite.tests:
test = test_class()
test_name = f"test_{suite.kind}_{test.name}"
setattr(empty_test_class, test_name, test.run)

def _with_stream_tests(
self,
empty_test_class: type[BaseTestClass],
suite: TestSuite[StreamTestTemplate],
streams: list[Stream],
) -> None:
params = [
{
"stream": stream,
}
for stream in streams
]
param_ids = [stream.name for stream in streams]

for test_class in suite.tests:
test = test_class()
test_name = f"test_{suite.kind}_{test.name}"
setattr(
empty_test_class,
test_name,
test.run,
)
empty_test_class.params[test_name] = params
empty_test_class.param_ids[test_name] = param_ids

def _with_stream_attribute_tests(
self,
empty_test_class: type[BaseTestClass],
suite: TestSuite[AttributeTestTemplate],
streams: list[Stream],
) -> None:
for test_class in suite.tests:
test = test_class()
test_name = f"test_{suite.kind}_{test.name}"
test_params = []
test_ids: list[str] = []
for stream in streams:
final_schema = stream.stream_maps[-1].transformed_schema["properties"]
test_params.extend(
[
{
"stream": stream,
"attribute_name": prop_name,
}
for stream in streams
]
param_ids = [stream.name for stream in streams]

for test_class in suite.tests:
test = test_class()
test_name = f"test_{suite.kind}_{test.name}"
setattr(
empty_test_class,
test_name,
test.run,
for prop_name, prop_schema in final_schema.items()
if test_class.evaluate(
stream=stream,
property_name=prop_name,
property_schema=prop_schema,
)
empty_test_class.params[test_name] = params
empty_test_class.param_ids[test_name] = param_ids

if suite.kind == "tap_stream_attribute":
for test_class in suite.tests:
test = test_class()
test_name = f"test_{suite.kind}_{test.name}"
test_params = []
test_ids: list[str] = []
for stream in streams:
final_schema = stream.stream_maps[-1].transformed_schema[
"properties"
]
test_params.extend(
[
{
"stream": stream,
"attribute_name": prop_name,
}
for prop_name, prop_schema in final_schema.items()
if test_class.evaluate(
stream=stream,
property_name=prop_name,
property_schema=prop_schema,
)
],
)
test_ids.extend(
[
f"{stream.name}.{prop_name}"
for prop_name, prop_schema in final_schema.items()
if test_class.evaluate(
stream=stream,
property_name=prop_name,
property_schema=prop_schema,
)
],
)

if test_params:
setattr(
empty_test_class,
test_name,
test.run,
)
empty_test_class.params[test_name] = test_params
empty_test_class.param_ids[test_name] = test_ids
],
)
test_ids.extend(
[
f"{stream.name}.{prop_name}"
for prop_name, prop_schema in final_schema.items()
if test_class.evaluate(
stream=stream,
property_name=prop_name,
property_schema=prop_schema,
)
],
)

return empty_test_class
if test_params:
setattr(
empty_test_class,
test_name,
test.run,
)
empty_test_class.params[test_name] = test_params
empty_test_class.param_ids[test_name] = test_ids


class TargetTestClassFactory:
Expand Down
8 changes: 4 additions & 4 deletions singer_sdk/testing/suites.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,17 +42,17 @@
TargetSchemaUpdates,
TargetSpecialCharsInAttributes,
)
from .templates import TestTemplate

if t.TYPE_CHECKING:
from .templates import TapTestTemplate, TargetTestTemplate, TestTemplate
T = t.TypeVar("T", bound=TestTemplate)


@dataclass
class TestSuite:
class TestSuite(t.Generic[T]):
"""Test Suite container class."""

kind: str
tests: list[type[TestTemplate] | type[TapTestTemplate] | type[TargetTestTemplate]]
tests: list[type[T]]


# Tap Test Suites
Expand Down