//go:build interop
package interop
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/identity"
"github.com/bluesky-social/indigo/atproto/syntax"
indigoxrpc "github.com/bluesky-social/indigo/xrpc"
"github.com/gorilla/websocket"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"tangled.org/core/knotfeed"
"tangled.org/core/tapc"
)
const seedRoute = "/xrpc/org.tangled.temp.server.seedRepo"
const fosterOwner = "did:plc:boltfulboy"
const fosterRkey = "the-feral-pearl-ledger"
const fosterCreatedAt = "1970-01-01T00:18:20Z"
func seedPost(t *testing.T, user, password string, body any) (int, string) {
t.Helper()
encoded, err := json.Marshal(body)
require.NoError(t, err)
request, err := http.NewRequest(
http.MethodPost,
os.Getenv("KNOT_TEST_BASE_URL")+seedRoute,
bytes.NewReader(encoded),
)
require.NoError(t, err)
request.Header.Set("Content-Type", "application/json")
if user != "" {
request.SetBasicAuth(user, password)
}
client := &http.Client{Timeout: 30 * time.Second}
response, err := client.Do(request)
require.NoError(t, err)
defer response.Body.Close()
raw, err := io.ReadAll(response.Body)
require.NoError(t, err)
return response.StatusCode, string(raw)
}
func seedRequest(t *testing.T) (map[string]any, syntax.RecordKey) {
t.Helper()
rkey, err := syntax.ParseRecordKey(fosterRkey)
require.NoError(t, err, "fosterRkey must be a record key")
fosterRepoDid := os.Getenv("KNOT_TEST_FOSTER_REPO_DID")
require.NotEmpty(
t,
fosterRepoDid,
"KNOT_TEST_FOSTER_REPO_DID is the foreign repo DID the knot fosters",
)
return map[string]any{
"repoDid": fosterRepoDid,
"manifest": map[string]any{
"$type": "org.tangled.repo.manifest",
"declaration": map[string]any{"owner": fosterOwner, "name": "boltfulboy"},
},
"records": []map[string]any{{
"collection": ticketCollection,
"rkey": rkey.String(),
"author": fosterOwner,
"createdAt": 1100,
"record": map[string]any{
"$type": ticketCollection,
"title": map[string]any{"$type": "org.tangled.markup.markdown#inline", "text": "a feral pearl"},
"state": "open",
"createdAt": fosterCreatedAt,
"x-tngl-editor": fosterOwner,
},
}},
}, rkey
}
func pushRepo(t *testing.T, source, addr, did, token string) {
t.Helper()
header := base64.StdEncoding.EncodeToString([]byte("x-tangled-token:" + token))
command := exec.Command(
"git", "-C", source,
"-c", "http.extraHeader=Authorization: Basic "+header,
"push", "-q", "--tags",
fmt.Sprintf("http://%s/%s", addr, did),
"main",
)
command.Env = append(os.Environ(), "GIT_TERMINAL_PROMPT=0")
output, err := command.CombinedOutput()
require.NoError(t, err, "the receive-pack push must deliver the staged history: %s", output)
}
func cloneRepo(t *testing.T, url string) string {
t.Helper()
clone := filepath.Join(t.TempDir(), "clone")
cloneCommand := exec.Command("git", "clone", "-q", url, clone)
cloneCommand.Env = append(os.Environ(), "GIT_TERMINAL_PROMPT=0")
output, err := cloneCommand.CombinedOutput()
require.NoError(t, err, "git clone must fetch the fostered repo: %s", output)
return clone
}
func gitIn(t *testing.T, dir string, args ...string) string {
t.Helper()
output, err := exec.Command("git", append([]string{"-C", dir}, args...)...).CombinedOutput()
require.NoError(t, err, "git %v failed: %s", args, output)
return strings.TrimSpace(string(output))
}
func TestASeededKnotFostersAForeignRepoVerbatim(t *testing.T) {
base := os.Getenv("KNOT_TEST_BASE_URL")
require.NotEmpty(t, base, "KNOT_TEST_BASE_URL is the knot under test")
addr := os.Getenv("KNOT_TEST_ADDR")
require.NotEmpty(t, addr, "KNOT_TEST_ADDR is the knot's host:port")
secret := os.Getenv("KNOT_TEST_FOSTER_SECRET")
require.NotEmpty(t, secret, "KNOT_TEST_FOSTER_SECRET is the foster secret")
source := os.Getenv("KNOT_TEST_FOSTER_SOURCE")
require.NotEmpty(t, source, "KNOT_TEST_FOSTER_SOURCE is the staged source repo")
wantSha := os.Getenv("KNOT_TEST_SHA")
require.NotEmpty(t, wantSha, "KNOT_TEST_SHA is the staged source tip")
pushToken := os.Getenv("KNOT_TEST_PUSH_TOKEN")
require.NotEmpty(t, pushToken, "KNOT_TEST_PUSH_TOKEN authorizes the receive-pack push")
ctx := context.Background()
xc := knotClient(t)
body, rkey := seedRequest(t)
wrongPasswordStatus, wrongPasswordBody := seedPost(t, "admin", "not-the-secret", body)
wrongUserStatus, wrongUserBody := seedPost(t, "root", secret, body)
noHeaderStatus, noHeaderBody := seedPost(t, "", "", body)
for _, attempt := range []struct {
status int
name string
}{
{wrongPasswordStatus, "a wrong password"},
{wrongUserStatus, "a wrong user"},
{noHeaderStatus, "no header"},
} {
require.Equal(t, http.StatusUnauthorized, attempt.status, "%s is rejected", attempt.name)
}
assert.Equal(t, wrongPasswordBody, wrongUserBody, "a wrong password and a wrong user are indistinguishable")
assert.Equal(t, wrongPasswordBody, noHeaderBody, "a missing header is indistinguishable too")
status, raw := seedPost(t, "admin", secret, body)
require.Equal(t, http.StatusOK, status, "the seed request must succeed: %s", raw)
var seeded struct {
Did string `json:"did"`
Key string `json:"key"`
Results []struct {
Collection string `json:"collection"`
Rkey string `json:"rkey"`
Error string `json:"error"`
} `json:"results"`
}
require.NoError(t, json.Unmarshal([]byte(raw), &seeded))
require.Equal(
t,
os.Getenv("KNOT_TEST_FOSTER_REPO_DID"),
seeded.Did,
"the response echoes the supplied repo DID, never a minted one",
)
require.NotEmpty(t, seeded.Key, "the response reports the repo's signing key")
require.Len(t, seeded.Results, 1, "one record was seeded")
assert.Equal(t, ticketCollection, seeded.Results[0].Collection)
assert.Equal(t, rkey.String(), seeded.Results[0].Rkey)
assert.Empty(t, seeded.Results[0].Error, "the verbatim record seeds without lexicon validation")
did, err := syntax.ParseDID(seeded.Did)
require.NoError(t, err)
collection, err := syntax.ParseNSID(ticketCollection)
require.NoError(t, err)
var listed struct {
Records []struct {
Uri string `json:"uri"`
Value struct {
State string `json:"state"`
CreatedAt string `json:"createdAt"`
Editor string `json:"x-tngl-editor"`
} `json:"value"`
} `json:"records"`
}
err = xc.LexDo(ctx, indigoxrpc.Query, "", "com.atproto.repo.listRecords", map[string]any{
"repo": did.String(),
"collection": ticketCollection,
"limit": 50,
}, nil, &listed)
require.NoError(t, err, "repo.listRecords serves the fostered repo")
require.Len(t, listed.Records, 1, "exactly one ticket is listed")
assert.Equal(t, "at://"+did.String()+"/"+ticketCollection+"/"+rkey.String(), listed.Records[0].Uri)
assert.Equal(t, "feral", listed.Records[0].Value.State, "the lexicon-invalid state is served verbatim")
assert.Equal(t, fosterCreatedAt, listed.Records[0].Value.CreatedAt, "the 1970 createdAt is served verbatim")
assert.Equal(t, fosterOwner, listed.Records[0].Value.Editor, "the foreign editor is served verbatim")
var fetched struct {
Uri string `json:"uri"`
Value struct {
Title struct {
Type string `json:"$type"`
Text string `json:"text"`
} `json:"title"`
} `json:"value"`
}
err = xc.LexDo(ctx, indigoxrpc.Query, "", "com.atproto.repo.getRecord", map[string]any{
"repo": did.String(),
"collection": ticketCollection,
"rkey": rkey.String(),
}, nil, &fetched)
require.NoError(t, err, "repo.getRecord serves the seeded record")
assert.Equal(t, listed.Records[0].Uri, fetched.Uri)
assert.Equal(t, "org.tangled.markup.markdown#inline", fetched.Value.Title.Type, "the title union is served in the org.tangled shape")
assert.Equal(t, "a feral pearl", fetched.Value.Title.Text)
document := fmt.Sprintf(
`{"id":%q,"verificationMethod":[{"id":%q,"type":"Multikey","controller":%q,"publicKeyMultibase":%q}]}`,
did.String(), did.String()+"#atproto", did.String(), seeded.Key,
)
var doc identity.DIDDocument
require.NoError(t, json.Unmarshal([]byte(document), &doc), "the seed response includes the key material of a did document")
dir := identity.NewMockDirectory()
dir.Insert(identity.ParseIdentity(&doc))
_, full, _ := provenRecord(t, ctx, xc, dir, did, collection, rkey)
leaf, _, err := full.GetRecordBytes(ctx, collection, rkey)
require.NoError(t, err, "the stored record bytes come back out of the signed tree")
stored, unions := leafFields(t, leaf)
assert.Equal(t, ticketCollection, stored["$type"])
assert.Equal(t, "feral", stored["state"], "the cbor leaf keeps the state the validated write path would refuse")
assert.Equal(t, fosterCreatedAt, stored["createdAt"], "the cbor leaf keeps the 1970 createdAt")
assert.Equal(t, fosterOwner, stored["x-tngl-editor"], "the cbor leaf keeps the foreign editor")
title, ok := unions["title"]
require.True(t, ok, "the leaf stores the title as the inline markup union")
assert.Equal(t, "org.tangled.markup.markdown#inline", title["$type"])
assert.Equal(t, "a feral pearl", title["text"])
pushRepo(t, source, addr, did.String(), pushToken)
clone := cloneRepo(t, "http://"+addr+"/"+did.String())
assert.Equal(t, wantSha, gitIn(t, clone, "rev-parse", "HEAD"), "the clone's HEAD is the pushed tip")
assert.Equal(t, gitIn(t, source, "symbolic-ref", "--short", "HEAD"), gitIn(t, clone, "symbolic-ref", "--short", "HEAD"), "the clone keeps the source's default branch")
assert.Equal(t, "v1.0.0", gitIn(t, clone, "tag", "--points-at", "HEAD"), "the annotated tag survives the round trip")
}
func TestFosteredReposAnnounceOnTheFirehose(t *testing.T) {
base := os.Getenv("KNOT_TEST_BASE_URL")
require.NotEmpty(t, base, "KNOT_TEST_BASE_URL is the knot under test")
addr := os.Getenv("KNOT_TEST_ADDR")
require.NotEmpty(t, addr, "KNOT_TEST_ADDR is the knot's host:port")
secret := os.Getenv("KNOT_TEST_FOSTER_SECRET")
require.NotEmpty(t, secret, "KNOT_TEST_FOSTER_SECRET is the foster secret")
source := os.Getenv("KNOT_TEST_FOSTER_SOURCE")
require.NotEmpty(t, source, "KNOT_TEST_FOSTER_SOURCE is the staged source repo")
wantSha := os.Getenv("KNOT_TEST_SHA")
require.NotEmpty(t, wantSha, "KNOT_TEST_SHA is the staged source tip")
body, rkey := seedRequest(t)
endpoint := "ws://" + strings.TrimPrefix(base, "http://") + "/xrpc/com.atproto.sync.subscribeRepos"
conn, _, err := websocket.DefaultDialer.DialContext(context.Background(), endpoint, nil)
require.NoError(t, err)
defer conn.Close()
status, raw := seedPost(t, "admin", secret, body)
require.Equal(t, http.StatusOK, status, "the seed request must succeed: %s", raw)
var seeded struct {
Did string `json:"did"`
}
require.NoError(t, json.Unmarshal([]byte(raw), &seeded))
did, err := syntax.ParseDID(seeded.Did)
require.NoError(t, err)
wantPath := ticketCollection + "/" + rkey.String()
announced := false
scan := time.Now().Add(15 * time.Second)
for !announced && time.Now().Before(scan) {
frame := nextFrame(t, conn)
if frame.Commit == nil || frame.Commit.Repo != did.String() {
continue
}
for _, op := range frame.Commit.Ops {
if op.Path != wantPath || op.Action != string(tapc.RecordCreateAction) {
continue
}
announced = true
assert.NotNil(t, op.Cid, "the live op cites the record cid")
assert.NotEmpty(t, frame.Commit.Blocks, "the live commit frame includes its CAR")
}
}
require.True(t, announced, "a live commit frame announced the seeded record")
pushToken := os.Getenv("KNOT_TEST_PUSH_TOKEN")
require.NotEmpty(t, pushToken, "KNOT_TEST_PUSH_TOKEN authorizes the receive-pack push")
pushRepo(t, source, addr, seeded.Did, pushToken)
records, cursor := knotfeedClient(t, addr, did)
sawRecord := false
sawRefs := false
deadline := time.After(60 * time.Second)
var settle <-chan time.Time
replay:
for {
if sawRecord && sawRefs && settle == nil {
settle = time.After(2 * time.Second)
}
select {
case op := <-records:
assert.NotEqual(t, tapc.RecordDeleteAction, op.Action,
"replay from zero mustn't serve a delete for %s/%s", op.Collection.String(), op.Rkey.String())
switch {
case op.Collection.String() == ticketCollection && op.Rkey == rkey:
assert.Equal(t, tapc.RecordCreateAction, op.Action, "the replayed seed op is a create")
assert.NotEmpty(t, op.Bytes, "the replayed op includes the record bytes")
sawRecord = true
case op.Collection.String() == "org.tangled.git.refs":
assert.Equal(t, "self", op.Rkey.String(), "the refs snapshot is a single self record")
def, refs, err := knotfeed.DecodeRefsSnapshot(op.Bytes)
require.NoError(t, err, "the frozen record envelope decodes")
if def != "main" {
break
}
for _, line := range strings.Split(string(refs), "\n") {
line = strings.TrimSpace(line)
sha, name, ok := strings.Cut(line, " ")
if !ok || name != "refs/heads/main" || sha != wantSha {
continue
}
sawRefs = true
}
}
case <-settle:
break replay
case <-deadline:
t.Fatalf("replay served the seeded record %v and the refs snapshot %v", sawRecord, sawRefs)
}
}
require.Eventually(t, func() bool { return cursor.Load() >= int64(1) },
5*time.Second, 10*time.Millisecond, "the consumer advanced its cursor")
}