notarius/internal/framework/checkpoint/recorder_test.go

package checkpoint

import (
	"os"
	"path/filepath"
	"testing"

	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)

func TestRootBasedRecorderOutputIsReusable(t *testing.T) {
	root := t.TempDir()
	identity := testIdentity(t)
	recorder, err := NewFilesystemRecorder(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	if err := recorder.ExtractSucceeded("lane", "module", nil, nil, nil, nil); err != nil {
		t.Fatal(err)
	}
	loader, err := NewFilesystemLoader(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	result, decision := loader.Extract("lane", "module", nil)
	if !decision.Reused || len(result.Outputs) != 0 {
		t.Fatalf("load result=%#v decision=%#v", result, decision)
	}
	relative, err := identity.RelativePath()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := os.Stat(filepath.Join(root, relative, "extract", "lane", "manifest.json")); err != nil {
		t.Fatal(err)
	}
}

func TestStepAwareRecorderAndLoaderIsolateLaneState(t *testing.T) {
	root := t.TempDir()
	identity := testIdentity(t)
	recorder, err := NewFilesystemRecorder(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	if err := recorder.(pipeline.StepCheckpointRecorder).ExtractSucceededForStep("step-a", "lane", "module", nil, nil, nil, nil); err != nil {
		t.Fatal(err)
	}
	loader, err := NewFilesystemLoader(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	stepLoader := loader.(pipeline.StepCheckpointLoader)
	if _, decision := stepLoader.ExtractForStep("step-b", "lane", "module", nil); decision.Reused {
		t.Fatal("checkpoint from another step was reused")
	}
	loaded, decision := stepLoader.ExtractForStep("step-a", "lane", "module", nil)
	if !decision.Reused || len(loaded.Outputs) != 0 {
		t.Fatalf("step-aware load = %#v, decision=%#v", loaded, decision)
	}
	relative, err := identity.RelativePath()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := os.Stat(filepath.Join(root, relative, "extract", "step-a", "lane", "manifest.json")); err != nil {
		t.Fatal(err)
	}
}

func TestStepAwareCheckpointPreservesDistinctDotIdentities(t *testing.T) {
	root := t.TempDir()
	identity := testIdentity(t)
	recorder, err := NewFilesystemRecorder(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	stepRecorder := recorder.(pipeline.StepCheckpointRecorder)
	for _, test := range []struct {
		stepID  string
		content string
	}{
		{stepID: ".", content: `{"identity":"dot"}`},
		{stepID: "..", content: `{"identity":"dot-dot"}`},
	} {
		artifact := pipeline.CheckpointArtifact{
			LaneID: "lane", ModuleKey: "normalize-module", SourceID: "source", ChunkID: "chunk", ChunkRef: source.SourceRef{SourceID: "source", StartUnitID: 1, EndUnitID: 1}, SchemaDigest: "sha256:schema",
			Artifact: contracts.SerializedArtifact{Kind: "kind", Schema: contracts.ArtifactSchema{ID: "schema", Name: "Schema", Version: "1"}, MediaType: "application/json", Content: []byte(test.content)},
		}
		if err := stepRecorder.NormalizeSucceededForStep(test.stepID, "lane", "normalize-module", nil, artifact, nil); err != nil {
			t.Fatalf("record %q: %v", test.stepID, err)
		}
	}

	loader, err := NewFilesystemLoader(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	for _, test := range []struct {
		stepID  string
		content string
		path    string
	}{
		{stepID: ".", content: `{"identity":"dot"}`, path: "%2E"},
		{stepID: "..", content: `{"identity":"dot-dot"}`, path: "%2E%2E"},
	} {
		loaded, decision := loader.AcceptedNormalize(test.stepID, "lane", "normalize-module")
		if !decision.Reused || string(loaded.Output.Artifact.Content) != test.content {
			t.Errorf("load %q = %#v, decision=%#v", test.stepID, loaded, decision)
		}
		relative, err := identity.RelativePath()
		if err != nil {
			t.Fatal(err)
		}
		if _, err := os.Stat(filepath.Join(root, relative, "normalize", test.path, "lane", "manifest.json")); err != nil {
			t.Errorf("checkpoint for %q: %v", test.stepID, err)
		}
	}
}

func TestCheckpointSchemaCompatibilityIdentifiers(t *testing.T) {
	if WorkspaceSchemaVersion != "notarius.workspace.v3" || WorkspaceSchemaVersionV2 != "notarius.workspace.v2" || WorkspaceSchemaVersionV1 != "notarius.workspace.v1" {
		t.Fatal("checkpoint schema identifiers are incorrect")
	}
}

func testIdentity(t *testing.T) Identity {
	t.Helper()
	identity, err := NewIdentity(IdentityInput{Pipeline: pipeline.ResolvedPipeline{ID: "pipeline", Digest: "sha256:aaaaaaaaaaaaaaaa", Input: pipeline.Binding("input")}, RawInputDigest: "sha256:bbbbbbbbbbbbbbbb"})
	if err != nil {
		t.Fatal(err)
	}
	return identity
}