You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
389 lines
14 KiB
389 lines
14 KiB
package mock
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"tera-clone/services/tera-api/internal/domain"
|
|
"tera-clone/services/tera-api/internal/logservice"
|
|
"tera-clone/services/tera-api/internal/logstorage"
|
|
)
|
|
|
|
type memoryJournal struct {
|
|
mu sync.Mutex
|
|
records map[string][]json.RawMessage
|
|
retentions []logstorage.Retention
|
|
snapshot logstorage.Snapshot
|
|
}
|
|
|
|
func (j *memoryJournal) Append(class string, _ time.Time, value any) error {
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
if j.records == nil {
|
|
j.records = make(map[string][]json.RawMessage)
|
|
}
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
j.records[class] = append(j.records[class], payload)
|
|
return nil
|
|
}
|
|
|
|
func (j *memoryJournal) Read(class string) ([]json.RawMessage, error) {
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
result := make([]json.RawMessage, len(j.records[class]))
|
|
for index := range result {
|
|
result[index] = append(json.RawMessage(nil), j.records[class][index]...)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (j *memoryJournal) ApplyRetention(retention logstorage.Retention) error {
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
j.retentions = append(j.retentions, retention)
|
|
return nil
|
|
}
|
|
|
|
func (j *memoryJournal) Snapshot() (logstorage.Snapshot, error) {
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
return j.snapshot, nil
|
|
}
|
|
|
|
func TestOperationalStoragePressureAlarmIsEdgeTriggered(t *testing.T) {
|
|
journal := &memoryJournal{snapshot: logstorage.Snapshot{
|
|
UsageBytes: map[string]int64{"audit": 20, "system": 30, "port": 31},
|
|
CapBytes: map[string]int64{"audit": 20, "system": 30, "port": 50},
|
|
}}
|
|
controller := &recordingAlarmDeliveryController{}
|
|
store, err := NewStoreWithOptions(fixedClock, StoreOptions{LogJournal: journal, AlarmDeliveryController: controller})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := store.EvaluateOperationalAlarms(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := store.EvaluateOperationalAlarms(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(controller.requests) != 1 || controller.requests[0].RuleID != "storage_pressure" {
|
|
t.Fatalf("storage pressure requests = %+v, want one edge-triggered alarm", controller.requests)
|
|
}
|
|
journal.mu.Lock()
|
|
journal.snapshot.UsageBytes["port"] = 20
|
|
journal.mu.Unlock()
|
|
if err := store.EvaluateOperationalAlarms(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
journal.mu.Lock()
|
|
journal.snapshot.UsageBytes["port"] = 31
|
|
journal.mu.Unlock()
|
|
if err := store.EvaluateOperationalAlarms(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(controller.requests) != 2 {
|
|
t.Fatalf("storage pressure did not rearm after recovery: %+v", controller.requests)
|
|
}
|
|
}
|
|
|
|
type sequencedMemoryJournal struct {
|
|
memoryJournal
|
|
sequences map[string][]uint64
|
|
}
|
|
|
|
func (j *sequencedMemoryJournal) Append(class string, timestamp time.Time, value any) error {
|
|
_, err := j.AppendSequence(class, timestamp, value)
|
|
return err
|
|
}
|
|
|
|
func (j *sequencedMemoryJournal) AppendSequence(class string, _ time.Time, value any) (uint64, error) {
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
if j.records == nil {
|
|
j.records = make(map[string][]json.RawMessage)
|
|
}
|
|
if j.sequences == nil {
|
|
j.sequences = make(map[string][]uint64)
|
|
}
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
sequence := uint64(len(j.sequences[class]) + 1)
|
|
j.records[class] = append(j.records[class], payload)
|
|
j.sequences[class] = append(j.sequences[class], sequence)
|
|
return sequence, nil
|
|
}
|
|
|
|
func (j *sequencedMemoryJournal) ReadAfter(class string, afterSequence uint64) ([]logservice.SequencedPayload, error) {
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
result := make([]logservice.SequencedPayload, 0)
|
|
for index, sequence := range j.sequences[class] {
|
|
if sequence > afterSequence {
|
|
result = append(result, logservice.SequencedPayload{Sequence: sequence, Payload: append(json.RawMessage(nil), j.records[class][index]...)})
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func TestSequencedJournalStartupLoadsOnlyConfiguredRecentEvents(t *testing.T) {
|
|
j := &sequencedMemoryJournal{}
|
|
now := time.Date(2026, 8, 22, 1, 2, 3, 0, time.UTC)
|
|
for sequence := 1; sequence <= 5; sequence++ {
|
|
event := domain.AuditEvent{
|
|
AuditID: fmt.Sprintf("audit_20260822_%06d", sequence),
|
|
Timestamp: now.Add(time.Duration(sequence) * time.Second),
|
|
Actor: "tester",
|
|
Action: "startup.test",
|
|
Target: "journal",
|
|
Result: "succeeded",
|
|
Message: fmt.Sprintf("event-%d", sequence),
|
|
}
|
|
if err := j.Append(logstorage.ClassAudit, event.Timestamp, event); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
j.snapshot.NextSequence = map[string]uint64{logstorage.ClassAudit: 6}
|
|
|
|
store, err := NewStoreWithOptions(func() time.Time { return now }, StoreOptions{
|
|
LogJournal: j,
|
|
AuditMaxEvents: 2,
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
events, err := store.AuditEvents(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(events) != 2 || events[0].Message != "event-4" || events[1].Message != "event-5" {
|
|
t.Fatalf("startup audit events = %+v, want only the newest two", events)
|
|
}
|
|
}
|
|
|
|
func TestSequencedJournalRefreshMergesExternalPortRecordsWithoutDuplicates(t *testing.T) {
|
|
journal := &sequencedMemoryJournal{}
|
|
now := time.Date(2026, 8, 4, 1, 2, 3, 0, time.UTC)
|
|
store, err := NewStoreWithOptions(func() time.Time { return now }, StoreOptions{LogJournal: journal})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := store.RecordPortLog(context.Background(), domain.PortLogEvent{Channel: "CH01", EventType: "traffic", Direction: "TX", Message: "local-1"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := journal.Append(logstorage.ClassPort, now, domain.PortLogEvent{Timestamp: now, Channel: "CH01", EventType: "traffic", Direction: "RX", Actor: "tera-seriald", Message: "external"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := store.RecordPortLog(context.Background(), domain.PortLogEvent{Channel: "CH01", EventType: "traffic", Direction: "TX", Message: "local-2"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
events, err := store.PortLogEvents(context.Background(), domain.PortLogFilter{Channel: "CH01"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(events) != 3 || events[0].Message != "local-1" || events[1].Message != "external" || events[2].Message != "local-2" {
|
|
t.Fatalf("merged port events = %+v", events)
|
|
}
|
|
events, err = store.PortLogEvents(context.Background(), domain.PortLogFilter{Channel: "CH01"})
|
|
if err != nil || len(events) != 3 {
|
|
t.Fatalf("repeated refresh events = %+v, %v", events, err)
|
|
}
|
|
}
|
|
|
|
func TestProductionJournalBacksRecordQueryAndRetentionApply(t *testing.T) {
|
|
journal := &memoryJournal{}
|
|
now := time.Date(2026, 7, 29, 1, 0, 0, 0, time.UTC)
|
|
store, err := NewStoreWithOptions(func() time.Time { return now }, StoreOptions{LogJournal: journal})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := store.RecordSystemLog(context.Background(), domain.SystemLogEvent{Severity: "info", Source: "test", Message: "durable"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
reloaded, err := NewStoreWithOptions(func() time.Time { return now }, StoreOptions{LogJournal: journal})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
events, err := reloaded.SystemLogEvents(context.Background(), domain.SystemLogFilter{})
|
|
if err != nil || len(events) != 1 || events[0].Message != "durable" {
|
|
t.Fatalf("SystemLogEvents = %+v, %v", events, err)
|
|
}
|
|
days := 7
|
|
if _, err := reloaded.UpdateLogRetention(context.Background(), domain.LogRetentionUpdate{SystemDays: &days}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := reloaded.ApplyConfig(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
journal.mu.Lock()
|
|
latest := journal.retentions[len(journal.retentions)-1]
|
|
journal.mu.Unlock()
|
|
if latest.SystemDays != 7 || latest.AuditDays != 365 || latest.PortDays != 365 {
|
|
t.Fatalf("latest retention = %+v", latest)
|
|
}
|
|
}
|
|
|
|
func TestProductionJournalRejectsPrototypePaths(t *testing.T) {
|
|
_, err := NewStoreWithOptions(time.Now, StoreOptions{LogJournal: &memoryJournal{}, AuditPath: "prototype.jsonl"})
|
|
if err == nil {
|
|
t.Fatal("NewStoreWithOptions accepted split-brain log persistence")
|
|
}
|
|
}
|
|
|
|
func TestProductionJournalBacksFilteredStreamingExport(t *testing.T) {
|
|
journal := &memoryJournal{}
|
|
store, err := NewStoreWithOptions(time.Now, StoreOptions{LogJournal: journal})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, event := range []domain.PortLogEvent{
|
|
{Channel: "CH01", EventType: "command", Command: "show LOGIN", ReturnValue: "ok", Message: "matched"},
|
|
{Channel: "CH02", EventType: "command", Command: "show LOGIN", ReturnValue: "ok", Message: "different channel"},
|
|
} {
|
|
if _, err := store.RecordPortLog(context.Background(), event); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
var output bytes.Buffer
|
|
operation, err := store.StreamLogExport(
|
|
domain.WithActor(context.Background(), domain.Actor{Name: "admin"}),
|
|
domain.LogExportRequest{LogType: "port", Channel: "CH01", Keyword: "login", MaxRows: 10},
|
|
&output,
|
|
)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if operation.State != "succeeded" || operation.AuditID == "" {
|
|
t.Fatalf("operation = %+v", operation)
|
|
}
|
|
lines := strings.Split(strings.TrimSpace(output.String()), "\n")
|
|
if len(lines) != 1 {
|
|
t.Fatalf("export line count = %d, want 1; output: %s", len(lines), output.String())
|
|
}
|
|
var event domain.PortLogEvent
|
|
if err := json.Unmarshal([]byte(lines[0]), &event); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if event.Channel != "CH01" || event.Command != "show LOGIN" {
|
|
t.Fatalf("event = %+v", event)
|
|
}
|
|
}
|
|
|
|
func TestProductionJournalStreamingExportSupportsWebPortFilters(t *testing.T) {
|
|
journal := &memoryJournal{}
|
|
store, err := NewStoreWithOptions(time.Now, StoreOptions{LogJournal: journal})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, event := range []domain.PortLogEvent{
|
|
{Channel: "CH25", EventType: "traffic", Direction: "RX", Message: "received"},
|
|
{Channel: "CH25", EventType: "traffic", Direction: "TX", Message: "sent"},
|
|
{Channel: "CH25", EventType: "command", Direction: "RX", Command: "show clock", ReturnValue: "ok", Message: "command"},
|
|
{Channel: "CH01", EventType: "traffic", Direction: "RX", Message: "other channel"},
|
|
} {
|
|
if _, err := store.RecordPortLog(context.Background(), event); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
var output bytes.Buffer
|
|
operation, err := store.StreamLogExport(
|
|
domain.WithActor(context.Background(), domain.Actor{Name: "admin"}),
|
|
domain.LogExportRequest{LogType: "port", Channel: "CH25", EventType: "traffic", Direction: "rx", MaxRows: 10},
|
|
&output,
|
|
)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if operation.Result["event_type"] != "traffic" || operation.Result["direction"] != "RX" {
|
|
t.Fatalf("operation filters = %+v", operation.Result)
|
|
}
|
|
lines := strings.Split(strings.TrimSpace(output.String()), "\n")
|
|
if len(lines) != 1 {
|
|
t.Fatalf("export line count = %d, want 1; output: %s", len(lines), output.String())
|
|
}
|
|
var event domain.PortLogEvent
|
|
if err := json.Unmarshal([]byte(lines[0]), &event); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if event.Channel != "CH25" || event.EventType != "traffic" || event.Direction != "RX" {
|
|
t.Fatalf("exported event = %+v", event)
|
|
}
|
|
}
|
|
|
|
func TestProductionJournalStreamingExportSupportsWebLinkFilters(t *testing.T) {
|
|
journal := &memoryJournal{}
|
|
now := time.Date(2026, 8, 15, 1, 2, 3, 0, time.UTC)
|
|
store, err := NewStoreWithOptions(func() time.Time { return now }, StoreOptions{LogJournal: journal})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, observation := range []domain.LinkStateObservation{
|
|
{Kind: "serial", Resource: "CH01", State: "up", Signal: "dcd", ObservedAt: now.Add(-time.Minute)},
|
|
{Kind: "ethernet", Resource: "Interface-1", State: "up", Signal: "carrier", ObservedAt: now.Add(-time.Minute)},
|
|
} {
|
|
if _, err := store.RecordLinkState(context.Background(), observation); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
var output bytes.Buffer
|
|
operation, err := store.StreamLogExport(
|
|
domain.WithActor(context.Background(), domain.Actor{Name: "admin"}),
|
|
domain.LogExportRequest{LogType: "link", Kind: "serial", Resource: "CH01", State: "up", EventType: "snapshot", Keyword: "dcd", MaxRows: 10},
|
|
&output,
|
|
)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if operation.Result["kind"] != "serial" || operation.Result["resource"] != "CH01" || operation.Result["state"] != "up" || operation.Result["event_type"] != "snapshot" {
|
|
t.Fatalf("operation filters = %+v", operation.Result)
|
|
}
|
|
lines := strings.Split(strings.TrimSpace(output.String()), "\n")
|
|
if len(lines) != 1 {
|
|
t.Fatalf("export line count = %d, want 1; output: %s", len(lines), output.String())
|
|
}
|
|
var event domain.LinkLogEvent
|
|
if err := json.Unmarshal([]byte(lines[0]), &event); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if event.Kind != "serial" || event.Resource != "CH01" || event.EventType != "snapshot" {
|
|
t.Fatalf("exported event = %+v", event)
|
|
}
|
|
}
|
|
|
|
func TestProductionJournalReadsCapturedSerialPayloadAndDirectionFilter(t *testing.T) {
|
|
journal := &memoryJournal{}
|
|
now := time.Date(2026, 7, 31, 2, 3, 4, 0, time.UTC)
|
|
want := []byte{0x00, 0xff, 'A', '\r', '\n'}
|
|
if err := journal.Append(logstorage.ClassPort, now, domain.PortLogEvent{
|
|
Timestamp: now, Channel: "CH01", EventType: "traffic", Direction: "RX", Actor: "session-1",
|
|
Message: "RX 5 serial payload byte(s)", Payload: want, PayloadBytes: len(want), PayloadEncoding: "base64",
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
store, err := NewStoreWithOptions(func() time.Time { return now }, StoreOptions{LogJournal: journal})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
events, err := store.PortLogEvents(context.Background(), domain.PortLogFilter{Channel: "CH01", Direction: "RX"})
|
|
if err != nil || len(events) != 1 || !bytes.Equal(events[0].Payload, want) || events[0].PayloadEncoding != "base64" {
|
|
t.Fatalf("captured events = %+v, %v", events, err)
|
|
}
|
|
events, err = store.PortLogEvents(context.Background(), domain.PortLogFilter{Direction: "TX"})
|
|
if err != nil || len(events) != 0 {
|
|
t.Fatalf("TX-filtered events = %+v, %v", events, err)
|
|
}
|
|
}
|
|
|