Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
# tests/test_pipeline.pyimport pytest
from pipeline import OrdersPipeline
from tests.fixtures import generate_orders_fixture
classTestOrdersPipeline:
@pytest.fixturedefpipeline(self, tmp_path):
return OrdersPipeline(
source_path=tmp_path / "source",
target_path=tmp_path / "target"
)
@pytest.fixturedefsource_data(self, tmp_path):
df = generate_orders_fixture(100)
path = tmp_path / "source" / "orders.csv"
path.parent.mkdir(parents=True)
df.to_csv(path, index=False)
return df
deftest_row_count_preserved(self, pipeline, source_data):
"""Verify no rows lost in transformation."""
pipeline.run()
result = pd.read_parquet(pipeline.target_path / "orders.parquet")
assertlen(result) == len(source_data)
deftest_all_columns_present(self, pipeline, source_data):
"""Verify output has expected columns."""
pipeline.run()
result = pd.read_parquet(pipeline.target_path / "orders.parquet")
expected_columns = ['order_id', 'customer_id', 'total', 'tier', 'processed_at']
assertall(col in result.columns for col in expected_columns)
deftest_no_null_required_fields(self, pipeline, source_data):
"""Verify required fields are populated."""
pipeline.run()
result = pd.read_parquet(pipeline.target_path / "orders.parquet")
assert result['order_id'].notna().all()
assert result['customer_id'].notna().all()
deftest_idempotent(self, pipeline, source_data):
"""Running twice produces same result."""
pipeline.run()
first_result = pd.read_parquet(pipeline.target_path / "orders.parquet")
pipeline.run()
second_result = pd.read_parquet(pipeline.target_path / "orders.parquet")
pd.testing.assert_frame_equal(first_result, second_result)
Data Quality Tests (dbt-style)
# tests/test_data_quality.pyimport pytest
from sqlalchemy import create_engine, text
@pytest.fixturedefdb_connection():
engine = create_engine("postgresql://...")
with engine.connect() as conn:
yield conn
classTestOrdersTable:
deftest_unique_order_id(self, db_connection):
result = db_connection.execute(text("""
SELECT order_id, COUNT(*) as cnt
FROM orders
GROUP BY order_id
HAVING COUNT(*) > 1
"""))
duplicates = result.fetchall()
assertlen(duplicates) == 0, f"Found duplicate order_ids: {duplicates[:5]}"deftest_valid_status(self, db_connection):
result = db_connection.execute(text("""
SELECT DISTINCT status
FROM orders
WHERE status NOT IN ('pending', 'confirmed', 'shipped', 'delivered', 'cancelled')
"""))
invalid = result.fetchall()
assertlen(invalid) == 0, f"Found invalid statuses: {invalid}"deftest_positive_amounts(self, db_connection):
result = db_connection.execute(text("""
SELECT COUNT(*) FROM orders WHERE total < 0
"""))
negative_count = result.scalar()
assert negative_count == 0, f"Found {negative_count} orders with negative totals"
Golden File Testing
deftest_transform_matches_golden(self):
"""Compare output to known-good result."""
input_df = pd.read_csv("tests/fixtures/input.csv")
expected = pd.read_csv("tests/golden/expected_output.csv")
result = transform(input_df)
pd.testing.assert_frame_equal(result, expected)
Snapshot Testing
deftest_schema_snapshot(self, snapshot):
"""Ensure schema hasn't changed unexpectedly."""
result = transform(input_df)
schema = {col: str(dtype) for col, dtype in result.dtypes.items()}
snapshot.assert_match(json.dumps(schema, indent=2), "schema.json")
Property-Based Testing
from hypothesis import given, strategies as st
@given(st.floats(min_value=0, max_value=1e9))deftest_total_always_positive(amount):
"""Total should never go negative."""
result = calculate_tax(amount)
assert result >= 0@given(st.lists(st.integers(min_value=1, max_value=100), min_size=1))deftest_sum_equals_parts(values):
"""Aggregation should equal sum of parts."""
df = pd.DataFrame({'amount': values})
result = aggregate(df)
assert result == sum(values)
