Skip to content

Commit

Permalink
refactor: Break up TapTestClassFactory._annotate_test_class into simp…
Browse files Browse the repository at this point in the history 
…ler methods
  • Loading branch information
@edgarrmondragon
edgarrmondragon committed Nov 22, 2023
1 parent 4cc52db commit 88ea277
Show file tree
Hide file tree
Showing 2 changed files with 97 additions and 71 deletions.
160 changes: 93 additions & 67 deletions singer_sdk/testing/factory.py
View file
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
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

0 comments on commit 88ea277

Please sign in to comment.