| 1 | package daemon |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 14 | "github.com/FelineStateMachine/tinyrelay/internal/replication" |
| 15 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 16 | ) |
| 17 | |
| 18 | func TestMountedHTTPPortableImportReportsFinalCountAndContent(t *testing.T) { |
| 19 | app, tenant := testTenant(t) |
| 20 | secret := strings.Repeat("0" , 63) + "1" |
| 21 | items := make([]event.Event, 2) |
| 22 | var body bytes.Buffer |
| 23 | for i := range items { |
| 24 | items[i] = event.Event{Kind: 1, CreatedAt: time.Now().Unix() + int64(i), Content: "portable-http-" + string(rune('a' +i))} |
| 25 | if err := event.Sign(&items[i], secret); err != nil { |
| 26 | t.Fatal(err) |
| 27 | } |
| 28 | raw, err := json.Marshal(items[i]) |
| 29 | if err != nil { |
| 30 | t.Fatal(err) |
| 31 | } |
| 32 | body.Write(raw) |
| 33 | body.WriteByte('\n' ) |
| 34 | } |
| 35 | res := mountedRequest(t, app, "PUT" , "/import" , body.Bytes(), "" , true) |
| 36 | if res.Code != 200 { |
| 37 | t.Fatalf("mounted import status=%d body=%s" , res.Code, res.Body.String()) |
| 38 | } |
| 39 | var response struct { |
| 40 | Result struct { |
| 41 | Imported int `json:"imported"` |
| 42 | } `json:"result"` |
| 43 | } |
| 44 | if err := json.Unmarshal(res.Body.Bytes(), &response); err != nil { |
| 45 | t.Fatal(err) |
| 46 | } |
| 47 | if response.Result.Imported != len(items) { |
| 48 | t.Fatalf("mounted import count=%d want=%d" , response.Result.Imported, len(items)) |
| 49 | } |
| 50 | for _, item := range items { |
| 51 | rows, err := tenant.store.Query(context.Background(), event.Filter{IDs: []string{item.ID}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}}) |
| 52 | if err != nil || len(rows.Events) != 1 || rows.Events[0].Content != item.Content { |
| 53 | t.Fatalf("imported content missing for %s: %d %v" , item.ID, len(rows.Events), err) |
| 54 | } |
| 55 | } |
| 56 | } |
| 57 | |
| 58 | func TestDaemonPortableImportJobResumesAfterRestart(t *testing.T) { |
| 59 | ctx := context.Background() |
| 60 | dir := t.TempDir() |
| 61 | secret := strings.Repeat("0" , 63) + "1" |
| 62 | owner, err := event.PublicKey(secret) |
| 63 | if err != nil { |
| 64 | t.Fatal(err) |
| 65 | } |
| 66 | config := Config{DataDir: dir, DefaultTenant: "main" } |
| 67 | app, err := New(ctx, config) |
| 68 | if err != nil { |
| 69 | t.Fatal(err) |
| 70 | } |
| 71 | meta, err := app.Create(ctx, CreateOptions{Name: "main" , Owner: owner, Template: "default" }) |
| 72 | if err != nil { |
| 73 | t.Fatal(err) |
| 74 | } |
| 75 | if err := app.Start(ctx, "http://relay.test" ); err != nil { |
| 76 | t.Fatal(err) |
| 77 | } |
| 78 | tenant := app.tenants[meta.ID] |
| 79 | tenant.workCancel() |
| 80 | tenant.workWG.Wait() |
| 81 | item := event.Event{Kind: 1, CreatedAt: time.Now().Unix(), Content: "portable import" } |
| 82 | if err := event.Sign(&item, secret); err != nil { |
| 83 | t.Fatal(err) |
| 84 | } |
| 85 | raw, err := json.Marshal(item) |
| 86 | if err != nil { |
| 87 | t.Fatal(err) |
| 88 | } |
| 89 | if err := os.WriteFile(filepath.Join(meta.Paths.Root, "resume.jsonl" ), append(raw, '\n' ), 0600); err != nil { |
| 90 | t.Fatal(err) |
| 91 | } |
| 92 | if err := tenant.replication.AddJob(ctx, replication.JobSpec{ID: "resume-import" , Kind: replication.JobImport, Relays: []string{"resume.jsonl" }}); err != nil { |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | if err := app.Close(ctx); err != nil { |
| 96 | t.Fatal(err) |
| 97 | } |
| 98 | |
| 99 | app, err = New(ctx, config) |
| 100 | if err != nil { |
| 101 | t.Fatal(err) |
| 102 | } |
| 103 | defer app.Close(ctx) |
| 104 | if err := app.Start(ctx, "http://relay.test" ); err != nil { |
| 105 | t.Fatal(err) |
| 106 | } |
| 107 | reloaded := app.tenants[meta.ID] |
| 108 | if !waitFor(t, 8*time.Second, func() bool { |
| 109 | statuses, statusErr := reloaded.replication.ListJobStatus(ctx) |
| 110 | return statusErr == nil && len(statuses) == 1 && statuses[0].Phase == "complete" |
| 111 | }) { |
| 112 | statuses, _ := reloaded.replication.ListJobStatus(ctx) |
| 113 | t.Fatalf("portable import did not resume: %#v" , statuses) |
| 114 | } |
| 115 | result, err := reloaded.store.Query(ctx, event.Filter{IDs: []string{item.ID}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}}) |
| 116 | if err != nil || len(result.Events) != 1 { |
| 117 | t.Fatalf("imported event missing after restart: %d %v" , len(result.Events), err) |
| 118 | } |
| 119 | } |