internal/transport/contract_test.go

Ref: Size: 4.4 KiB History
package transport

import (
	"bytes"
	"testing"

	"github.com/a73x/eitri/internal/pb"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"google.golang.org/protobuf/encoding/protowire"
	"google.golang.org/protobuf/proto"
)

// The wire contract (internal/pb + internal/transport) is the only code shared
// across the control/data-plane boundary (invariant R3). These tests pin its
// observable behavior: every field of the two top-level envelopes must survive
// a WriteMsg/ReadMsg round-trip unchanged. If a field stops being framed, or
// the framing is altered incompatibly, proto.Equal fails here.

func TestAgentMessageReportRoundTrip(t *testing.T) {
	in := &pb.AgentMessage{Msg: &pb.AgentMessage_Report{Report: &pb.Report{
		Vms: []*pb.VMStatus{
			{VmId: "vm-1", PowerState: "on", Phase: "running", Ip: "10.0.0.5", LastError: ""},
			{VmId: "vm-2", PowerState: "off", Phase: "stopped"},
			{VmId: "vm-3", PowerState: "off", Phase: "creating", StatusDetail: "downloading image 1.2/3.7 GiB"},
		},
		Exposures: []*pb.ExposureStatus{{
			Id: "e-1", State: "active",
			Sessions: &pb.ExposureSessions{Active: 7, Refused: 12, Dropped: 3},
		}},
		Destroyed:      []string{"vm-old"},
		Quarantined:    []*pb.QuarantinedVM{{VmId: "vm-q", Name: "q", VmspecJson: []byte(`{"k":1}`), DestroyAtUnix: 1750000000}},
		Capacity:       &pb.Capacity{Vcpus: 16, MemMb: 32768, DiskGb: 500},
		FenceViolation: true,
		LastSeenEpoch:  42,
	}}}

	out := &pb.AgentMessage{}
	roundTrip(t, in, out)
	assert.True(t, proto.Equal(in, out), "report did not survive round-trip:\n in=%v\nout=%v", in, out)
}

func TestServerMessageSnapshotRoundTrip(t *testing.T) {
	in := &pb.ServerMessage{Msg: &pb.ServerMessage_Snapshot{Snapshot: &pb.Snapshot{
		Epoch: 7,
		Vms: []*pb.VMSpec{{
			VmId: "vm-1", Name: "web", ImageUrl: "https://img/x.qcow2", ImageSha256: "abc",
			CloudInit: "#cloud-config", Vcpus: 4, MemMb: 8192, DiskGb: 40,
			Persistent: true, PowerState: "on", Tombstoned: false, SshAuthorizedKey: "ssh-ed25519 AAAA",
			SshUserCaAuthorizedKeys: []string{"ssh-ed25519 CAAAAA eitri-user-ca"},
		}},
	}}}

	out := &pb.ServerMessage{}
	roundTrip(t, in, out)
	assert.True(t, proto.Equal(in, out), "snapshot did not survive round-trip:\n in=%v\nout=%v", in, out)
}

// TestAReportFromANewerAgentDecodesOnAnOlderServer pins the rule the field
// numbering exists to serve. A mixed fleet is the normal state during a
// rollout, so a peer that has never heard of a field must skip past it and read
// everything else — not reject the frame, and not misread the field it does
// know. Field 99 is nothing in this schema and stands in for whatever a later
// release adds beside status_detail.
func TestAReportFromANewerAgentDecodesOnAnOlderServer(t *testing.T) {
	raw, err := proto.Marshal(&pb.VMStatus{
		VmId: "vm-1", PowerState: "off", Phase: "creating",
		StatusDetail: "downloading image 1.2/3.7 GiB",
	})
	require.NoError(t, err)
	raw = protowire.AppendTag(raw, 99, protowire.BytesType)
	raw = protowire.AppendString(raw, "a field from a release this peer predates")

	var got pb.VMStatus
	require.NoError(t, proto.Unmarshal(raw, &got), "an unknown field must not fail the frame")
	assert.Equal(t, "vm-1", got.GetVmId())
	assert.Equal(t, "creating", got.GetPhase())
	assert.Equal(t, "downloading image 1.2/3.7 GiB", got.GetStatusDetail())
}

// TestAReportFromAnOlderAgentReadsAsSilence is the same rule from the other
// side, and the one the console's display rules rest on: an agent that predates
// these fields sends neither, and neither may arrive as a value. An absent
// status_detail is empty (nothing to add) and absent counters are NIL — not a
// zeroed struct claiming this port has turned nobody away.
func TestAReportFromAnOlderAgentReadsAsSilence(t *testing.T) {
	raw, err := proto.Marshal(&pb.Report{
		Vms:       []*pb.VMStatus{{VmId: "vm-1", PowerState: "off", Phase: "creating"}},
		Exposures: []*pb.ExposureStatus{{Id: "e-1", State: "active"}},
	})
	require.NoError(t, err)

	var got pb.Report
	require.NoError(t, proto.Unmarshal(raw, &got))
	assert.Empty(t, got.GetVms()[0].GetStatusDetail())
	assert.Nil(t, got.GetExposures()[0].GetSessions(), "an uncounted port must not decode as zeros")
}

// roundTrip frames in and reads it back into out.
func roundTrip(t *testing.T, in, out proto.Message) {
	t.Helper()
	var buf bytes.Buffer
	require.NoError(t, WriteMsg(&buf, in))
	require.NoError(t, ReadMsg(&buf, out, DefaultMaxFrame))
}