Netpulse_SasS/server/internal/httpapi/api_test.go

// Наскрізний тест API: справжня БД, справжній HTTP, справжній WebSocket.
//
//	NETPULSE_TEST_DSN="postgres://netpulse:netpulse@localhost/netpulse_it" go test ./...
//
// БАЗА МАЄ БУТИ ПОРОЖНЬОЮ — лише зі схемою, без даних.
//
// Кабінети тут унікальні (slug із наносекундами), а от КОРИСТУВАЧІ в
// цьому продукті глобальні: одна людина може працювати в кількох
// кабінетах, тому `core.users.username` унікальний наскрізно, і
// CreateUser при зіткненні свідомо підхоплює наявного, НЕ переписуючи
// пароль (users.go, і це правильно: інакше адмін одного кабінету міняв
// би пароль людини в чужому).
//
// Наслідок для тестів: на базі, де вже стоїть робоча інсталяція,
// seedUser("admin@example.com") натрапляє на власника з ім'ям `admin`,
// пароль лишається чужим, вхід не вдається — і тест падає на 401 у
// зовсім іншому місці. Це не поломка продукту, а невиконана умова.
//
// Так знайшлось: набір прогнали проти пісочниці установника, тобто
// проти ВЖЕ поставленої системи. CI і `./netpulse sandbox once` дають
// чисту базу, тому там усе зелене.
package httpapi_test

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"testing"
	"time"

	"bytes"

	"github.com/coder/websocket"
	"github.com/jackc/pgx/v5/pgxpool"
	npv1 "github.com/netpulse/netpulse/gen/go/netpulse/v1"
	"github.com/netpulse/netpulse/server/internal/auth"
	"github.com/netpulse/netpulse/server/internal/httpapi"
	"github.com/netpulse/netpulse/server/internal/store"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type fixture struct {
	pool  *pgxpool.Pool
	store *store.Store
	srv   *httptest.Server
	ctx   context.Context

	tenantID string
	token    string
	agentID  string
	deviceID string
	peerID   string
	ifaceID  string
	peerIfID string
	linkID   string
	mapID    string
	nodeID   string
}

func setup(t *testing.T) *fixture {
	t.Helper()

	dsn := os.Getenv("NETPULSE_TEST_DSN")
	if dsn == "" {
		t.Skip("NETPULSE_TEST_DSN не задано — інтеграційний тест пропущено")
	}

	ctx := context.Background()
	st, err := store.New(ctx, dsn)
	if err != nil {
		t.Fatalf("БД: %v", err)
	}
	t.Cleanup(st.Close)

	f := &fixture{pool: st.Pool(), store: st, ctx: ctx}
	f.seed(t)

	signer, err := auth.NewSigner([]byte("test-secret-at-least-32-bytes-long!!"))
	if err != nil {
		t.Fatalf("signer: %v", err)
	}
	api := httpapi.New(st, signer, slog.New(slog.NewTextHandler(io.Discard, nil)))

	hubCtx, cancel := context.WithCancel(ctx)
	go api.Hub().Run(hubCtx)
	t.Cleanup(cancel)

	f.srv = httptest.NewServer(api.Handler())
	t.Cleanup(f.srv.Close)

	return f
}

func (f *fixture) seed(t *testing.T) {
	t.Helper()
	ctx := f.ctx

	slug := fmt.Sprintf("api-%d", time.Now().UnixNano())
	f.token = "np_api_" + slug
	sum := sha256.Sum256([]byte(f.token))
	agentToken := sha256.Sum256([]byte("agent_" + slug))

	scan := func(dest *string, q string, args ...any) {
		t.Helper()
		if err := f.pool.QueryRow(ctx, q, args...).Scan(dest); err != nil {
			t.Fatalf("seed %q: %v", strings.TrimSpace(q)[:40], err)
		}
	}
	exec := func(q string, args ...any) {
		t.Helper()
		if _, err := f.pool.Exec(ctx, q, args...); err != nil {
			t.Fatalf("seed %q: %v", strings.TrimSpace(q)[:40], err)
		}
	}

	scan(&f.tenantID, `INSERT INTO core.tenants (slug, name, status)
		VALUES ($1, $2, 'active') RETURNING id::text`, slug, slug)

	t.Cleanup(func() {
		bg := context.Background()
		_, _ = f.pool.Exec(bg, `DELETE FROM ts.icmp_samples WHERE tenant_id = $1`, f.tenantID)
		_, _ = f.pool.Exec(bg, `DELETE FROM ts.if_counters WHERE tenant_id = $1`, f.tenantID)
		_, _ = f.pool.Exec(bg, `DELETE FROM ts.device_status_history WHERE tenant_id = $1`, f.tenantID)
		_, _ = f.pool.Exec(bg, `DELETE FROM core.tenants WHERE id = $1`, f.tenantID)
	})

	scan(&f.agentID, `INSERT INTO core.agents (tenant_id, name, token_hash, status)
		VALUES ($1, 'probe-api', $2, 'online') RETURNING id::text`, f.tenantID, agentToken[:])

	exec(`INSERT INTO core.api_tokens (tenant_id, name, prefix, token_hash, scopes)
	      VALUES ($1, 'ui', 'np_api_', $2, '{}')`, f.tenantID, sum[:])

	scan(&f.deviceID, `INSERT INTO inv.devices (tenant_id, agent_id, name, address, kind, status)
		VALUES ($1, $2, 'core-sw', '10.20.0.1', 'switch', 'up') RETURNING id::text`,
		f.tenantID, f.agentID)
	scan(&f.peerID, `INSERT INTO inv.devices (tenant_id, agent_id, name, address, kind, status)
		VALUES ($1, $2, 'edge-rtr', '10.20.0.2', 'router', 'up') RETURNING id::text`,
		f.tenantID, f.agentID)

	scan(&f.ifaceID, `INSERT INTO inv.interfaces (tenant_id, device_id, if_index, name, speed_bps, oper_status)
		VALUES ($1, $2, 1, 'Gi0/1', 1000000000, 'up') RETURNING id::text`, f.tenantID, f.deviceID)
	scan(&f.peerIfID, `INSERT INTO inv.interfaces (tenant_id, device_id, if_index, name, speed_bps, oper_status)
		VALUES ($1, $2, 1, 'ether1', 1000000000, 'up') RETURNING id::text`, f.tenantID, f.peerID)

	scan(&f.linkID, `INSERT INTO topo.links
		(tenant_id, a_device_id, a_interface_id, b_device_id, b_interface_id,
		 kind, capacity_bps, discovered_by, status)
		VALUES ($1,$2,$3,$4,$5,'physical',1000000000,'lldp','up') RETURNING id::text`,
		f.tenantID, f.deviceID, f.ifaceID, f.peerID, f.peerIfID)

	scan(&f.mapID, `INSERT INTO topo.maps (tenant_id, name, slug, kind)
		VALUES ($1, 'NOC', 'noc', 'logical') RETURNING id::text`, f.tenantID)

	exec(`INSERT INTO topo.map_backgrounds
		(tenant_id, map_id, kind, storage_key, mime_type, width, height, opacity)
		VALUES ($1,$2,'image','s3://plan.svg','image/svg+xml',2400,1600,0.6)`,
		f.tenantID, f.mapID)

	scan(&f.nodeID, `INSERT INTO topo.map_nodes (tenant_id, map_id, kind, device_id, label, x, y)
		VALUES ($1,$2,'device',$3,'core-sw',100,200) RETURNING id::text`,
		f.tenantID, f.mapID, f.deviceID)

	var peerNode string
	scan(&peerNode, `INSERT INTO topo.map_nodes (tenant_id, map_id, kind, device_id, label, x, y)
		VALUES ($1,$2,'device',$3,'edge-rtr',400,200) RETURNING id::text`,
		f.tenantID, f.mapID, f.peerID)

	exec(`INSERT INTO topo.map_edges
		(tenant_id, map_id, source_node_id, target_node_id,
		 source_interface_id, target_interface_id, link_id, label)
		VALUES ($1,$2,$3,$4,$5,$6,$7,'Gi0/1 → ether1')`,
		f.tenantID, f.mapID, f.nodeID, peerNode, f.ifaceID, f.peerIfID, f.linkID)

	// Живі лічильники: без них у ребра не буде util_pct, тобто нічим
	// керувати анімацією.
	exec(`INSERT INTO ts.if_counters
		(ts, interface_id, device_id, tenant_id, in_bps, out_bps, util_in_pct, util_out_pct, oper_up)
		VALUES (now(), $1, $2, $3, 420e6, 780e6, 42, 78, true)`,
		f.ifaceID, f.deviceID, f.tenantID)

	exec(`INSERT INTO ts.icmp_samples
		(ts, device_id, tenant_id, agent_id, rtt_avg_ms, loss_pct, packets_sent, packets_recv, reachable)
		VALUES (now(), $1, $2, $3, 1.5, 0, 3, 3, true)`,
		f.deviceID, f.tenantID, f.agentID)
}

// bytesReader — дрібний помічник для тіл запитів у тестах входу.
func bytesReader(b []byte) *bytes.Reader { return bytes.NewReader(b) }

func mustHash(t *testing.T, pw string) string {
	t.Helper()
	h, err := auth.HashPassword(pw)
	if err != nil {
		t.Fatalf("hash: %v", err)
	}
	return h
}

func sha256sum(s string) []byte {
	sum := sha256.Sum256([]byte(s))
	return sum[:]
}

func (f *fixture) get(t *testing.T, path, token string) (int, []byte) {
	t.Helper()
	req, err := http.NewRequest(http.MethodGet, f.srv.URL+path, nil)
	if err != nil {
		t.Fatalf("запит: %v", err)
	}
	if token != "" {
		req.Header.Set("Authorization", "Bearer "+token)
	}
	resp, err := f.srv.Client().Do(req)
	if err != nil {
		t.Fatalf("виклик %s: %v", path, err)
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	return resp.StatusCode, body
}

// ---------------------------------------------------------------------

func TestAuthRequired(t *testing.T) {
	f := setup(t)

	if code, _ := f.get(t, "/api/v1/maps", ""); code != http.StatusUnauthorized {
		t.Fatalf("без токена код %d", code)
	}
	if code, _ := f.get(t, "/api/v1/maps", "np_api_нема-такого"); code != http.StatusUnauthorized {
		t.Fatalf("з чужим токеном код %d", code)
	}
	// healthz свідомо відкритий: його опитує балансувальник.
	if code, _ := f.get(t, "/healthz", ""); code != http.StatusOK {
		t.Fatalf("healthz код %d", code)
	}
}

func TestListMaps(t *testing.T) {
	f := setup(t)

	code, body := f.get(t, "/api/v1/maps", f.token)
	if code != http.StatusOK {
		t.Fatalf("код %d: %s", code, body)
	}

	var out struct {
		Maps []store.MapSummary `json:"maps"`
	}
	if err := json.Unmarshal(body, &out); err != nil {
		t.Fatalf("розбір: %v", err)
	}
	if len(out.Maps) != 1 {
		t.Fatalf("мап: %d", len(out.Maps))
	}
	m := out.Maps[0]
	if m.Name != "NOC" || m.NodeCount != 2 || m.EdgeCount != 1 {
		t.Fatalf("зведення мапи неправильне: %+v", m)
	}
}

// Головний запит продукту: одним викликом фронтенд має отримати готове
// до рендеру полотно разом із живими статусами — інакше мапа малювалася б
// сірою й лише потім доганяла кольори сотнею дозапитів.
func TestMapStateIsRenderReady(t *testing.T) {
	f := setup(t)

	code, body := f.get(t, "/api/v1/maps/"+f.mapID, f.token)
	if code != http.StatusOK {
		t.Fatalf("код %d: %s", code, body)
	}

	var st store.MapState
	if err := json.Unmarshal(body, &st); err != nil {
		t.Fatalf("розбір: %v", err)
	}

	if st.Name != "NOC" || st.LayoutAlgo == "" {
		t.Fatalf("мапа: %+v", st)
	}
	if len(st.Viewport) == 0 || len(st.Grid) == 0 {
		t.Fatal("немає viewport/grid — полотно нема куди відновити")
	}

	if len(st.Backgrounds) != 1 || st.Backgrounds[0].StorageKey == "" {
		t.Fatalf("підкладки: %+v", st.Backgrounds)
	}
	if st.Backgrounds[0].Opacity != 0.6 {
		t.Fatalf("прозорість підкладки: %v", st.Backgrounds[0].Opacity)
	}

	if len(st.Nodes) != 2 {
		t.Fatalf("вузлів: %d", len(st.Nodes))
	}
	var found *store.MapNode
	for i := range st.Nodes {
		if st.Nodes[i].DeviceID == f.deviceID {
			found = &st.Nodes[i]
		}
	}
	if found == nil {
		t.Fatal("вузол пристрою не повернувся")
	}
	if found.X != 100 || found.Y != 200 {
		t.Fatalf("координати вузла: %v,%v", found.X, found.Y)
	}
	// Саме це фарбує вузол.
	if found.Status != "up" {
		t.Fatalf("статус вузла: %q", found.Status)
	}
	if found.RttMs == nil || *found.RttMs < 1.4 || *found.RttMs > 1.6 {
		t.Fatalf("RTT не підмішався: %v", found.RttMs)
	}

	if len(st.Edges) != 1 {
		t.Fatalf("ребер: %d", len(st.Edges))
	}
	e := st.Edges[0]
	if e.SourcePort != "Gi0/1" || e.TargetPort != "ether1" {
		t.Fatalf("порти ребра: %q → %q", e.SourcePort, e.TargetPort)
	}
	// Стан лінка виводиться з кінців, а не читається з колонки:
	// обидва пристрої up і порти up → лінк up.
	if e.LinkStatus != "up" {
		t.Fatalf("статус лінка: %q", e.LinkStatus)
	}
	// Джерело швидкості анімації.
	if e.UtilPct == nil || *e.UtilPct != 78 {
		t.Fatalf("util_pct: %v", e.UtilPct)
	}
	if e.CapacityBps == nil || *e.CapacityBps != 1000000000 {
		t.Fatalf("capacity_bps: %v", e.CapacityBps)
	}
	if len(e.Animation) == 0 || len(e.Thresholds) == 0 {
		t.Fatal("немає налаштувань анімації/порогів")
	}
}

// Стан лінка — похідна від його кінців. Колонка topo.links.status
// ніким не підтримується, тому читати її означало б завжди показувати
// "unknown" і брехати про обрив.
func TestLinkStatusFollowsEndpoints(t *testing.T) {
	f := setup(t)

	edgeStatus := func() string {
		t.Helper()
		code, body := f.get(t, "/api/v1/maps/"+f.mapID, f.token)
		if code != http.StatusOK {
			t.Fatalf("код %d: %s", code, body)
		}
		var st store.MapState
		if err := json.Unmarshal(body, &st); err != nil {
			t.Fatalf("розбір: %v", err)
		}
		if len(st.Edges) != 1 {
			t.Fatalf("ребер: %d", len(st.Edges))
		}
		return st.Edges[0].LinkStatus
	}

	if got := edgeStatus(); got != "up" {
		t.Fatalf("обидва кінці живі, а лінк %q", got)
	}

	// Один кінець упав — лінк має почервоніти, хоча в topo.links
	// нічого не змінювалось.
	if _, err := f.pool.Exec(f.ctx,
		`UPDATE inv.devices SET status = 'down' WHERE id = $1`, f.peerID); err != nil {
		t.Fatalf("зміна статусу: %v", err)
	}
	if got := edgeStatus(); got != "down" {
		t.Fatalf("кінець лежить, а лінк %q", got)
	}

	// Порт адміністративно вимкнено — теж обрив.
	if _, err := f.pool.Exec(f.ctx,
		`UPDATE inv.devices SET status = 'up' WHERE id = $1`, f.peerID); err != nil {
		t.Fatalf("відновлення: %v", err)
	}
	if _, err := f.pool.Exec(f.ctx,
		`UPDATE inv.interfaces SET oper_status = 'down' WHERE id = $1`, f.ifaceID); err != nil {
		t.Fatalf("зміна порту: %v", err)
	}
	if got := edgeStatus(); got != "down" {
		t.Fatalf("порт лежить, а лінк %q", got)
	}
}

// Токен одного тенанта не має відкривати мапу іншого навіть за точним id.
func TestTenantIsolation(t *testing.T) {
	f := setup(t)
	other := setup(t)

	code, body := f.get(t, "/api/v1/maps/"+other.mapID, f.token)
	if code != http.StatusNotFound {
		t.Fatalf("чужа мапа віддалась із кодом %d: %s", code, body)
	}

	code, body = f.get(t, "/api/v1/devices", f.token)
	if code != http.StatusOK {
		t.Fatalf("код %d", code)
	}
	if strings.Contains(string(body), other.deviceID) {
		t.Fatal("у переліку пристроїв видно чужий тенант")
	}
}

func TestBadMapID(t *testing.T) {
	f := setup(t)

	if code, _ := f.get(t, "/api/v1/maps/не-uuid", f.token); code != http.StatusBadRequest {
		t.Fatalf("некоректний id дав код %d, очікували 400", code)
	}
}

func TestListDevicesAndAgents(t *testing.T) {
	f := setup(t)

	code, body := f.get(t, "/api/v1/devices", f.token)
	if code != http.StatusOK {
		t.Fatalf("код %d", code)
	}
	var dev struct {
		Devices []store.DeviceSummary `json:"devices"`
	}
	if err := json.Unmarshal(body, &dev); err != nil {
		t.Fatalf("розбір: %v", err)
	}
	if len(dev.Devices) != 2 {
		t.Fatalf("пристроїв: %d", len(dev.Devices))
	}
	for _, d := range dev.Devices {
		if d.Name == "core-sw" && d.IfaceCount != 1 {
			t.Fatalf("лічильник інтерфейсів: %d", d.IfaceCount)
		}
	}

	code, body = f.get(t, "/api/v1/agents", f.token)
	if code != http.StatusOK {
		t.Fatalf("код %d", code)
	}
	var ag struct {
		Agents []store.AgentSummary `json:"agents"`
	}
	if err := json.Unmarshal(body, &ag); err != nil {
		t.Fatalf("розбір: %v", err)
	}
	if len(ag.Agents) != 1 || ag.Agents[0].DeviceCount != 2 {
		t.Fatalf("зонди: %+v", ag.Agents)
	}
}

// ---------------------------------------------------------------------
// WebSocket
// ---------------------------------------------------------------------

func dialWS(t *testing.T, ctx context.Context, f *fixture, token string) *websocket.Conn {
	t.Helper()

	url := "ws" + strings.TrimPrefix(f.srv.URL, "http") + "/api/v1/ws"
	conn, _, err := websocket.Dial(ctx, url, &websocket.DialOptions{
		HTTPClient: f.srv.Client(),
		// Браузерний WebSocket не дозволяє довільні заголовки, тому
		// токен їде підпротоколом — перевіряємо саме цей шлях.
		Subprotocols: []string{"netpulse.token." + token},
	})
	if err != nil {
		t.Fatalf("WebSocket: %v", err)
	}
	t.Cleanup(func() { conn.CloseNow() })
	return conn
}

func readMsg(t *testing.T, ctx context.Context, conn *websocket.Conn) map[string]any {
	t.Helper()
	_, data, err := conn.Read(ctx)
	if err != nil {
		t.Fatalf("читання WebSocket: %v", err)
	}
	var m map[string]any
	if err := json.Unmarshal(data, &m); err != nil {
		t.Fatalf("розбір повідомлення: %v", err)
	}
	return m
}

// waitFor читає, доки не побачить повідомлення потрібного типу.
func waitFor(t *testing.T, ctx context.Context, conn *websocket.Conn, want string) map[string]any {
	t.Helper()
	for {
		m := readMsg(t, ctx, conn)
		if m["type"] == want {
			return m
		}
		if m["type"] == "error" {
			t.Fatalf("сервер повернув помилку: %v", m)
		}
	}
}

func TestWebSocketRequiresToken(t *testing.T) {
	f := setup(t)

	ctx, cancel := context.WithTimeout(f.ctx, 10*time.Second)
	defer cancel()

	url := "ws" + strings.TrimPrefix(f.srv.URL, "http") + "/api/v1/ws"
	if conn, _, err := websocket.Dial(ctx, url, &websocket.DialOptions{
		HTTPClient: f.srv.Client(),
	}); err == nil {
		conn.CloseNow()
		t.Fatal("WebSocket відкрився без токена")
	}
}

// Зміна статусу пристрою має долетіти до браузера сама — заради цього
// й існує вся зв'язка outbox → hub → WebSocket.
func TestWebSocketDeliversStatusChange(t *testing.T) {
	f := setup(t)

	ctx, cancel := context.WithTimeout(f.ctx, 30*time.Second)
	defer cancel()

	conn := dialWS(t, ctx, f, f.token)
	waitFor(t, ctx, conn, "hello")

	if err := conn.Write(ctx, websocket.MessageText,
		[]byte(`{"type":"subscribe","map_id":"`+f.mapID+`"}`)); err != nil {
		t.Fatalf("підписка: %v", err)
	}
	sub := waitFor(t, ctx, conn, "subscribed")
	if sub["map_id"] != f.mapID {
		t.Fatalf("підписка на іншу мапу: %v", sub)
	}

	// Проганяємо справжній шлях: батч телеметрії з недосяжним
	// пристроєм → зміна статусу → подія в outbox → трансляція.
	agent := &store.Agent{ID: f.agentID, TenantID: f.tenantID}
	table := store.NewSeriesTable()
	if _, err := f.store.WriteBatch(ctx, agent, &npv1.TelemetryBatch{
		BatchId: 1, AgentId: f.agentID,
		Icmp: []*npv1.IcmpResult{{
			DeviceId: f.deviceID, Ts: timestamppb.Now(),
			LossPct: 100, PacketsSent: 3, PacketsRecv: 0, Reachable: false,
		}},
	}, table); err != nil {
		t.Fatalf("запис батчу: %v", err)
	}

	msg := waitFor(t, ctx, conn, "device.status")
	payload, ok := msg["payload"].(map[string]any)
	if !ok {
		t.Fatalf("подія без корисного навантаження: %v", msg)
	}
	if payload["device_id"] != f.deviceID {
		t.Fatalf("подія про інший пристрій: %v", payload)
	}
	if payload["status"] != "down" {
		t.Fatalf("статус у події: %v", payload["status"])
	}
	if payload["previous_status"] != "up" {
		t.Fatalf("попередній статус: %v", payload["previous_status"])
	}
}

// Підписка на чужу мапу за вгаданим id не має відкривати топологію.
func TestWebSocketRejectsForeignMap(t *testing.T) {
	f := setup(t)
	other := setup(t)

	ctx, cancel := context.WithTimeout(f.ctx, 15*time.Second)
	defer cancel()

	conn := dialWS(t, ctx, f, f.token)
	waitFor(t, ctx, conn, "hello")

	if err := conn.Write(ctx, websocket.MessageText,
		[]byte(`{"type":"subscribe","map_id":"`+other.mapID+`"}`)); err != nil {
		t.Fatalf("підписка: %v", err)
	}

	m := readMsg(t, ctx, conn)
	if m["type"] != "error" || m["code"] != "map_not_found" {
		t.Fatalf("чужа мапа прийнята до підписки: %v", m)
	}
}

// Завантаження каналів штовхається періодично — це те, що рухає
// анімацію без перемальовування полотна.
func TestWebSocketPushesLinkLoads(t *testing.T) {
	f := setup(t)

	ctx, cancel := context.WithTimeout(f.ctx, 30*time.Second)
	defer cancel()

	conn := dialWS(t, ctx, f, f.token)
	waitFor(t, ctx, conn, "hello")

	if err := conn.Write(ctx, websocket.MessageText,
		[]byte(`{"type":"subscribe","map_id":"`+f.mapID+`"}`)); err != nil {
		t.Fatalf("підписка: %v", err)
	}
	waitFor(t, ctx, conn, "subscribed")

	msg := waitFor(t, ctx, conn, "link.load")
	if msg["map_id"] != f.mapID {
		t.Fatalf("оновлення для іншої мапи: %v", msg)
	}

	links, ok := msg["links"].([]any)
	if !ok || len(links) == 0 {
		t.Fatalf("немає лінків у оновленні: %v", msg)
	}
	first, _ := links[0].(map[string]any)
	if first["link_id"] != f.linkID {
		t.Fatalf("інший лінк: %v", first)
	}
	if util, ok := first["util_pct"].(float64); !ok || util != 78 {
		t.Fatalf("завантаження не доїхало: %v", first["util_pct"])
	}
}