Subclass the matching test harness in your plugin's test suite; pytest auto-discovers every test_* method. You get tens of conformance tests for free.
PluginTestHarness (universal — covers any BasePlugin)
├── InfraProviderTestHarness (adds: role == provider, base is InfraProvider)
├── CustomScaffoldTestHarness (adds: determinism, idempotency, path-traversal, ...)
├── ValidatorTestHarness (adds: finding severity, base is Validator)
└── CatalogAdapterTestHarness (adds: role == catalog, base is CatalogAdapter)
# tests/test_conformance.py
from fluid_sdk.testing import CustomScaffoldTestHarness, LOCAL_CONTRACT
from my_pkg.scaffold import MyCIScaffold
class TestMyCIScaffold(CustomScaffoldTestHarness):
plugin_class = MyCIScaffold
init_kwargs = {} # any extra ctor kwargs
sample_contracts = [LOCAL_CONTRACT]Run with pytest tests/test_conformance.py -v. Every inherited test_* method runs against your plugin.
- Plugin subclasses
BasePlugin. nameis a non-empty identifier matching^[a-z][a-z0-9_\-]*$.nameis not in a reserved list (unknown,base,test, ...).roleis declared (non-empty class attribute).- Constructor accepts
init_kwargs. loggeris set after construction.capabilities()returns aMapping[str, bool]withplanningandapplykeys.plan()returns a list.- Every action in
plan()has bothopandresource_id. plan()is deterministic — same contract twice → same list.get_plugin_info()returns metadata whosenamematches the class.- (Opt-in)
apply()returns anExecutionResult-like object with the expected fields.
All of the above, plus:
- Subclass declares
role = "custom_scaffold". - Subclass is a
CustomScaffoldsubclass. - Every action's
opis eitherwrite_fileorskip. - Every
write_fileaction carriesparams.path+params.content_b64. apply()writes the expected number of files to a tempdir underoutput_root.- Re-applying the same actions produces byte-identical output (idempotency).
- No action's
params.pathis absolute or contains..(path-traversal safety).
All of PluginTestHarness, plus:
- Subclass declares
role = "provider". - Subclass is an
InfraProvidersubclass.
All of PluginTestHarness, plus:
- Subclass declares
role = "validator". - Subclass is a
Validatorsubclass. applyis not skipped (validators are cheap to run).- Every
emit_findingaction hasparams.severityin{info, warn, error, critical}.
All of PluginTestHarness, plus:
- Subclass declares
role = "catalog". - Subclass is a
CatalogAdaptersubclass.
The SDK ships two fixtures (under fluid_sdk.testing.fixtures):
MINIMAL_CONTRACT— sparse. Exercises that plugins don't crash on empty exposes/builds/consumes.LOCAL_CONTRACT— slightly richer. Has one expose, one build, one environment, governance labels.
Plugin authors should usually add their own sample_contracts covering domain-specific shapes:
class TestMyProvider(InfraProviderTestHarness):
plugin_class = MyProvider
sample_contracts = [
LOCAL_CONTRACT, # the SDK baseline
{ # plugin-specific fixture
"fluidVersion": "0.7.4",
"id": "my-test-product",
"exposes": [{...}],
...
},
]Default skip_apply = True because most cloud-infra apply() impls hit real cloud APIs. Override to False for plugins that are safe to apply in tests:
class TestMyScaffold(CustomScaffoldTestHarness):
plugin_class = MyScaffold
sample_contracts = [LOCAL_CONTRACT]
# CustomScaffoldTestHarness already has skip_apply implicitly False
# (scaffolds write to tempdir — always safe in tests)Use pytest's standard mechanisms — pytest.mark.skip or override the method in your subclass:
class TestMyScaffold(CustomScaffoldTestHarness):
plugin_class = MyScaffold
sample_contracts = [LOCAL_CONTRACT]
@pytest.mark.skip(reason="Plugin uses random IDs by design — non-deterministic")
def test_plan_is_deterministic(self):
pass(Don't do this lightly. Determinism is a strong contract.)
Just add more test_* methods to the subclass; pytest discovers them automatically.
class TestMyScaffold(CustomScaffoldTestHarness):
plugin_class = MyScaffold
sample_contracts = [LOCAL_CONTRACT]
def test_my_specific_invariant(self):
scaffold = self.get_plugin()
actions = scaffold.plan(LOCAL_CONTRACT)
# custom assertion
assert any(a["resource_id"] == "README.md" for a in actions)Run conformance tests in CI alongside your normal test suite:
# .github/workflows/test.yml
- name: Run conformance tests
run: pytest tests/ -v --tb=shortIf your plugin advertises support for many platforms / engines, parametrise:
@pytest.mark.parametrize("contract", [LOCAL_CONTRACT, CUSTOM_FIXTURE_1, CUSTOM_FIXTURE_2])
class TestMyScaffold(CustomScaffoldTestHarness):
plugin_class = MyScaffold
@property
def sample_contracts(self):
return [self.contract] # pytest injects via parametrize