carlos / platform Public

Download this file

1	// End-to-end acceptance for `carlos logs` (#131/#78): a REAL console.Server
2	// over a DirStore holding shipped log chunks, a device-flow login, then the
3	// real runLogs driven in-process — the member's whole path from "my app is
4	// 503ing" to reading the wake failure, with no box access anywhere.
5	// TestE2EInstancesThroughConsole's shape.
6	package main
7
8	import (
9	"context"
10	"net/http"
11	"net/http/httptest"
12	"path/filepath"
13	"strings"
14	"testing"
15	"time"
16
17	"github.com/carlosframework/platform/internal/applog"
18	"github.com/carlosframework/platform/internal/console"
19	"github.com/carlosframework/platform/internal/release"
20	)
21
22	func TestE2ELogsThroughConsole(t *testing.T) {
23	useTempCredentials(t)
24	clearDeploymentEnv(t)
25	checkNoRegistry := requireNoRegistryTouch(t)
26
27	tmp := t.TempDir()
28	relStore := release.NewDirStore(filepath.Join(tmp, "store"))
29	seed, pubB64, err := release.GenerateKey()
30	if err != nil {
31	t.Fatal(err)
32	}
33	priv, err := release.ParseSigningKey(seed, "e2e")
34	if err != nil {
35	t.Fatal(err)
36	}
37	pub, err := release.ParsePublicKey(pubB64, "e2e")
38	if err != nil {
39	t.Fatal(err)
40	}
41	cst, err := console.Open(filepath.Join(tmp, "console.db"))
42	if err != nil {
43	t.Fatal(err)
44	}
45	t.Cleanup(func() { cst.Close() })
46
47	var srv *console.Server
48	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
49	srv.ServeHTTP(w, r)
50	}))
51	t.Cleanup(ts.Close)
52	srv, err = console.NewServer(console.Config{
53	Origin: ts.URL,
54	Keymail: "http://keymail.invalid", // never dialled — see e2eDeviceLogin
55	Registry: filepath.Join(tmp, "registry.db"),
56	Releases: relStore,
57	ReleasePub: pub,
58	Key: priv,
59	AppScheme: "http",
60	AppsDomain: "oncarlos.test",
61	}, cst)
62	if err != nil {
63	t.Fatal(err)
64	}
65
66	const addr = "paul@keymail.dev"
67	if err := cst.AddMember(console.OpsAccountID, addr, console.RoleOwner, time.Now()); err != nil {
68	t.Fatal(err)
69	}
70	if _, err := cst.CreateAppIn(console.OpsAccountID, "titogo", addr, time.Now()); err != nil {
71	t.Fatal(err)
72	}
73	token, expiresAt := e2eDeviceLogin(t, ts.URL, cst, addr, "e2e@laptop")
74	credFile, err := loadCredentials()
75	if err != nil {
76	t.Fatal(err)
77	}
78	credFile.put(credential{Console: ts.URL, Address: addr, Token: token, ExpiresAt: expiresAt})
79	if err := credFile.save(); err != nil {
80	t.Fatal(err)
81	}
82
83	// A box shipped two chunks: routine output, then a wake failure — the
84	// #131 case, as the shipping loop would have written it.
85	now := time.Now().UTC()
86	writeChunk := func(seq uint64, recs []applog.Record) {
87	t.Helper()
88	body, err := applog.EncodeChunk(recs)
89	if err != nil {
90	t.Fatal(err)
91	}
92	key, err := applog.ChunkKey(release.OpsAccount, "titogo", "box1", applog.HourOf(recs[0].TS), seq, len(body))
93	if err != nil {
94	t.Fatal(err)
95	}
96	if _, err := relStore.Put(context.Background(), key, body, ""); err != nil {
97	t.Fatal(err)
98	}
99	}
100	writeChunk(100, []applog.Record{
101	{TS: now.Add(-20 * time.Minute), Box: "box1", Host: "titogo.bab.oncarlos.test",
102	Stream: applog.StreamApp, Msg: "listening on socket", Seq: 100,
103	Fields: map[string]string{"request_id": "r1"}},
104	})
105	writeChunk(101, []applog.Record{
106	{TS: now.Add(-2 * time.Minute), Box: "box1", Host: "titogo.bab.oncarlos.test",
107	Stream: applog.StreamPlatform, Level: applog.LevelError, Seq: 101,
108	Msg: "wake failed: nothing is promoted for this instance's channel"},
109	})
110
111	// The plain read: both streams merged, oldest first, wake reason visible.
112	buf := captureStdout(t)
113	runLogs([]string{"-app", "titogo", "-account", "ops"})
114	out := buf.String()
115	if !strings.Contains(out, "listening on socket") || !strings.Contains(out, "wake failed") {
116	t.Fatalf("logs stdout missing the timeline:\n%s", out)
117	}
118	if strings.Index(out, "listening on socket") > strings.Index(out, "wake failed") {
119	t.Fatalf("timeline out of order:\n%s", out)
120	}
121	if !strings.Contains(out, "[platform]") || !strings.Contains(out, "[app]") {
122	t.Fatalf("stream tags missing:\n%s", out)
123	}
124	if !strings.Contains(out, "request_id=r1") {
125	t.Fatalf("structured fields missing:\n%s", out)
126	}
127
128	// Filters reach the wire: -stream platform drops the app line.
129	buf.Reset()
130	runLogs([]string{"-app", "titogo", "-account", "ops", "-stream", "platform"})
131	if out := buf.String(); strings.Contains(out, "listening on socket") || !strings.Contains(out, "wake failed") {
132	t.Fatalf("--stream platform: %q", out)
133	}
134
135	// -f follows: a line shipped between polls arrives; the poll cap is the
136	// test's exit.
137	logsMaxPolls = 2
138	t.Cleanup(func() { logsMaxPolls = 0 })
139	done := make(chan string, 1)
140	go func() {
141	fbuf := captureStdout(t)
142	runLogs([]string{"-app", "titogo", "-account", "ops", "-f"})
143	done <- fbuf.String()
144	}()
145	time.Sleep(500 * time.Millisecond) // after the initial read, before poll 1 lands
146	writeChunk(102, []applog.Record{
147	{TS: now.Add(-time.Second), Box: "box1", Host: "titogo.bab.oncarlos.test",
148	Stream: applog.StreamApp, Msg: "fresh follow line", Seq: 102},
149	})
150	select {
151	case out := <-done:
152	if !strings.Contains(out, "fresh follow line") {
153	t.Fatalf("follow missed the new line:\n%s", out)
154	}
155	case <-time.After(15 * time.Second):
156	t.Fatal("follow did not finish its bounded polls")
157	}
158
159	// The whole flow ran through the console: nothing here may have touched
160	// a local registry.
161	checkNoRegistry()
162	}
163