fix(storage): read canonical backup inventory

Canonical resource queries could return no backups because inventory compared
raw guest IDs from provider caches. Read org-scoped recovery artifacts with
shared identity, repository, verification and provenance fields instead.

Preserve global pagination across providers and surface read failures. Remove
the obsolete vendor summaries and unrelated recent-task cache from this read.
This commit is contained in:
rcourtman
2026-10-15 17:16:32 +01:00
parent 0e5b335b23
commit 5866248ff4
10 changed files with 250 additions and 288 deletions
+1 -1
View File
@@ -9376,7 +9376,7 @@
"lane_followups": [
{
"id": "architecture-post-rc-canonicalization",
"summary": "Track the remaining post-RC architecture cleanup that still belongs to L6 after canonicalizing Connected infrastructure and the frontend state contract by construction, including retirement of newly legacy or clearly obsolete old-way internal paths unless a boundary-only exception is explicitly governed. The investigation backup inventory overview still filters legacy guest identifiers (including a rune conversion for PVE VM IDs), so canonical resource queries can return empty while recovery points exist. Move that overview onto canonical recovery-point reads; backup-task failure reads and storage history parity do not close this separate inventory gap.",
"summary": "Track the remaining post-RC architecture cleanup that still belongs to L6 after canonicalizing Connected infrastructure and the frontend state contract by construction, including retirement of newly legacy or clearly obsolete old-way internal paths unless a boundary-only exception is explicitly governed.",
"owner": "project-owner",
"status": "planned",
"recorded_at": "2026-03-13",
@@ -36,7 +36,13 @@ The API and tool adapter share `Manager.ListPointsForOrg`, including the
canonical graph source in demo mode, rather than presenting fixture evidence
through HTTP alone while tools consult an empty store.
`TestInvestigationReadsCanonicalBackupFailureWithoutAgent` uses mapped, persisted
task evidence and checks subject and tenant isolation.
task evidence and checks subject and tenant isolation. The `backups` inventory
also reads this org-scoped source with `KindBackup` and canonical subject IDs.
It returns canonical points with repository references, provenance and optional
verification fields, rather than VM-ID-filtered PBS/PVE summary lists or an
unscoped recent-task cache. Global offset pagination preserves source ordering
across providers and does not scan earlier pages. Read failures remain errors;
an artifact record alone is not proof of restorability.
Storage performance reads resolve the canonical registry metrics target and use
the monitor-owned history selector. Responses contain usage percentages with
@@ -27,8 +27,12 @@ reader. Demo composition reads the canonical fixture graph through the recovery
query helpers; ordinary tenants read their persisted store. Moving the existing
in-memory filters into recovery preserves point, series and facet query semantics
and pagination bounds. The API/tool parity regression checks that a collected
backup failure remains readable without a PVE client or command agent. No route,
entitlement, recovery mutation or action approval is added.
backup failure remains readable without a PVE client or command agent. The
investigation backup inventory uses the same `KindBackup` and canonical subject
query as the points endpoint, retaining point identity, repository references,
tri-state verification and evidence provenance in real and demo modes. Tool
pagination translates global offsets into the shared source's bounded pages.
No route, entitlement, recovery mutation or action approval is added.
### Patrol setup panel reuses existing AI endpoints
+13 -23
View File
@@ -5,10 +5,10 @@ import (
"encoding/json"
"strings"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/recovery"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
@@ -22,34 +22,24 @@ func mustParseJSONMapControl(t *testing.T, text string) map[string]interface{} {
}
func TestPulseToolExecutor_ExecuteListBackups(t *testing.T) {
backupProv := &mockBackupProvider{}
resourceID := unifiedresources.ProxmoxGuestCanonicalID(unifiedresources.ResourceTypeVM, "west", 101)
point := recovery.RecoveryPoint{ID: "backup-artifact", Provider: recovery.ProviderProxmoxPBS, Kind: recovery.KindBackup, SubjectResourceID: resourceID}
exec := NewPulseToolExecutor(ExecutorConfig{
BackupProvider: backupProv,
ControlLevel: ControlLevelReadOnly,
RecoveryPointsProvider: inventoryPointsProviderFunc(func(ctx context.Context, opts recovery.ListPointsOptions) ([]recovery.RecoveryPoint, int, error) {
assert.Equal(t, recovery.KindBackup, opts.Kind)
assert.Equal(t, resourceID, opts.SubjectResourceID)
return []recovery.RecoveryPoint{point}, 1, nil
}),
ControlLevel: ControlLevelReadOnly,
})
expectedBackups := models.Backups{
PBS: []models.PBSBackup{
{VMID: "101", BackupType: "vm", BackupTime: time.Now(), Instance: "pbs1", Datastore: "ds1", Size: 1024 * 1024 * 1024},
},
PVE: models.PVEBackups{
StorageBackups: []models.StorageBackup{
{VMID: 102, Time: time.Now(), Size: 2 * 1024 * 1024 * 1024, Storage: "local"},
},
BackupTasks: []models.BackupTask{
{VMID: 101, Node: "node1", Status: "OK", StartTime: time.Now()},
},
},
}
backupProv.On("GetBackups").Return(expectedBackups)
backupProv.On("GetPBSInstances").Return([]models.PBSInstance{})
// Use pulse_storage tool with type: "backups"
result, err := exec.ExecuteTool(context.Background(), "pulse_storage", map[string]interface{}{
"type": "backups",
"type": "backups", "resource_id": resourceID,
})
assert.NoError(t, err)
assert.False(t, result.IsError)
var response BackupsResponse
assert.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &response))
assert.Equal(t, []recovery.RecoveryPoint{point}, response.Points)
}
func TestPulseToolExecutor_ExecuteControlGuest(t *testing.T) {
+8 -70
View File
@@ -3,6 +3,7 @@ package tools
import (
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/recovery"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/pkg/agents/filesystem"
)
@@ -1028,13 +1029,12 @@ func (r PatternsResponse) NormalizeCollections() PatternsResponse {
return r
}
// BackupsResponse is returned by pulse_list_backups
// BackupsResponse is the canonical artifact inventory returned by pulse_storage backups.
// RecoveryPoint preserves provider identity, repository references and unknown
// verification instead of collapsing them into vendor-specific summary lists.
type BackupsResponse struct {
PBS []PBSBackupSummary `json:"pbs"`
PVE []PVEBackupSummary `json:"pve"`
PBSServers []PBSServerSummary `json:"pbs_servers"`
RecentTasks []BackupTaskSummary `json:"recent_tasks"`
Pagination *PaginationInfo `json:"pagination,omitempty"`
Points []recovery.RecoveryPoint `json:"points"`
Pagination *PaginationInfo `json:"pagination"`
}
func EmptyBackupsResponse() BackupsResponse {
@@ -1042,74 +1042,12 @@ func EmptyBackupsResponse() BackupsResponse {
}
func (r BackupsResponse) NormalizeCollections() BackupsResponse {
if r.PBS == nil {
r.PBS = []PBSBackupSummary{}
}
if r.PVE == nil {
r.PVE = []PVEBackupSummary{}
}
if r.PBSServers == nil {
r.PBSServers = []PBSServerSummary{}
}
for i := range r.PBSServers {
r.PBSServers[i] = r.PBSServers[i].NormalizeCollections()
}
if r.RecentTasks == nil {
r.RecentTasks = []BackupTaskSummary{}
if r.Points == nil {
r.Points = []recovery.RecoveryPoint{}
}
return r
}
// PBSBackupSummary is a summarized PBS backup
type PBSBackupSummary struct {
VMID string `json:"vmid"`
BackupType string `json:"backup_type"`
BackupTime time.Time `json:"backup_time"`
Instance string `json:"instance"`
Datastore string `json:"datastore"`
SizeGB float64 `json:"size_gb"`
Verified bool `json:"verified"`
Protected bool `json:"protected"`
}
// PVEBackupSummary is a summarized PVE backup
type PVEBackupSummary struct {
VMID int `json:"vmid"`
BackupTime time.Time `json:"backup_time"`
SizeGB float64 `json:"size_gb"`
Storage string `json:"storage"`
}
// PBSServerSummary is a summarized PBS server
type PBSServerSummary struct {
Name string `json:"name"`
Host string `json:"host"`
Status string `json:"status"`
Datastores []DatastoreSummary `json:"datastores"`
}
func (s PBSServerSummary) NormalizeCollections() PBSServerSummary {
if s.Datastores == nil {
s.Datastores = []DatastoreSummary{}
}
return s
}
// DatastoreSummary is a summarized datastore
type DatastoreSummary struct {
Name string `json:"name"`
UsagePercent float64 `json:"usage_percent"`
FreeGB float64 `json:"free_gb"`
}
// BackupTaskSummary is a summarized backup task
type BackupTaskSummary struct {
VMID int `json:"vmid"`
Node string `json:"node"`
Status string `json:"status"`
StartTime time.Time `json:"start_time"`
}
// StorageResponse is returned by pulse_list_storage
type StorageResponse struct {
Pools []StoragePoolSummary `json:"pools"`
@@ -472,21 +472,8 @@ func TestSharedToolSummaryOwnersUseCanonicalEmptyCollections(t *testing.T) {
})
}
payload, err := json.Marshal(BackupsResponse{
PBSServers: []PBSServerSummary{{Name: "pbs-1"}},
}.NormalizeCollections())
require.NoError(t, err)
var decoded map[string]any
require.NoError(t, json.Unmarshal(payload, &decoded))
servers := decoded["pbs_servers"].([]any)
server := servers[0].(map[string]any)
datastores, ok := server["datastores"].([]any)
if !ok || len(datastores) != 0 {
t.Fatalf("expected pbs_servers[0].datastores to be an empty array, got %T (%v)", server["datastores"], server["datastores"])
}
payload, err = json.Marshal(StorageResponse{
payload, err := json.Marshal(StorageResponse{
Pools: []StoragePoolSummary{{ID: "pool-1", Name: "local"}},
}.NormalizeCollections())
require.NoError(t, err)
@@ -3,6 +3,7 @@ package tools
import (
"context"
"encoding/json"
"fmt"
"path/filepath"
"testing"
"time"
@@ -17,6 +18,128 @@ import (
"github.com/stretchr/testify/require"
)
func TestInvestigationBackupInventoryUsesCanonicalArtifacts(t *testing.T) {
ctx := context.Background()
manager := recoverymanager.New(config.NewMultiTenantPersistence(t.TempDir()))
store, err := manager.StoreForOrg("default")
require.NoError(t, err)
t.Cleanup(func() { _ = store.Close() })
at := time.Unix(1800, 0).UTC()
resourceID := unifiedresources.ProxmoxGuestCanonicalID(unifiedresources.ResourceTypeVM, "west", 101)
otherResourceID := unifiedresources.ProxmoxGuestCanonicalID(unifiedresources.ResourceTypeVM, "east", 101)
archives := []models.StorageBackup{
{ID: "west-archive", VMID: 101, Instance: "west", Node: "pve1", Type: "qemu", Storage: "local", Time: at, Size: 1024},
{ID: "east-archive", VMID: 101, Instance: "east", Node: "pve1", Type: "qemu", Storage: "local", Time: at, Size: 2048},
}
points := proxmoxmapper.FromPVEStorageBackups(archives, map[string]proxmoxmapper.GuestInfo{
"west|pve1|101": {ResourceID: resourceID, ResourceType: unifiedresources.ResourceTypeVM},
"east|pve1|101": {ResourceID: otherResourceID, ResourceType: unifiedresources.ResourceTypeVM},
})
points = append(points, proxmoxmapper.FromPBSBackups([]models.PBSBackup{{
ID: "remote-archive", VMID: "101", BackupType: "vm", Instance: "pbs-west", Datastore: "daily", BackupTime: at, Size: 4096,
}}, map[string][]proxmoxmapper.GuestCandidate{
"vm:101": {{ResourceID: resourceID, ResourceType: unifiedresources.ResourceTypeVM, InstanceName: "west", NodeName: "pve1", VMID: 101, SourceID: "west:pve1:101"}},
})...)
// Inventory is provider-agnostic and excludes execution/snapshot evidence.
points = append(points,
recovery.RecoveryPoint{ID: "agent-backup", Provider: recovery.ProviderAgent, Kind: recovery.KindBackup, Mode: recovery.ModeRemote, Outcome: recovery.OutcomeUnknown, SubjectResourceID: resourceID},
recovery.RecoveryPoint{ID: "pve-task:failure", Provider: recovery.ProviderProxmoxPVE, Kind: recovery.KindOther, SubjectResourceID: resourceID},
recovery.RecoveryPoint{ID: "pve-snapshot:old", Provider: recovery.ProviderProxmoxPVE, Kind: recovery.KindSnapshot, SubjectResourceID: resourceID},
)
require.NoError(t, store.UpsertPoints(ctx, points))
foreign, err := manager.StoreForOrg("foreign")
require.NoError(t, err)
t.Cleanup(func() { _ = foreign.Close() })
foreignPoint := points[0]
foreignPoint.ID = "foreign-archive"
require.NoError(t, foreign.UpsertPoints(ctx, []recovery.RecoveryPoint{foreignPoint}))
executor := NewPulseToolExecutor(ExecutorConfig{
RecoveryPointsProvider: NewRecoveryPointsToolAdapter(manager, "default"),
// A legacy cache must not shadow, supplement or filter canonical records.
BackupProvider: &stubBackupProvider{backups: models.Backups{PVE: models.PVEBackups{StorageBackups: archives}}},
})
result, err := executor.executeListBackups(ctx, map[string]interface{}{"resource_id": resourceID})
require.NoError(t, err)
require.False(t, result.IsError, result.Content)
var response struct {
Points []recovery.RecoveryPoint `json:"points"`
Pagination PaginationInfo `json:"pagination"`
}
require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &response))
require.Len(t, response.Points, 3)
require.Equal(t, 3, response.Pagination.Total)
byID := map[string]recovery.RecoveryPoint{}
for _, point := range response.Points {
require.Equal(t, resourceID, point.SubjectResourceID)
require.Equal(t, recovery.KindBackup, point.Kind)
byID[point.ID] = point
}
require.Contains(t, byID, "pve-backup:west-archive")
require.Nil(t, byID["pve-backup:west-archive"].Verified, "uncollected verification remains unknown")
require.Contains(t, byID, "pbs-backup:remote-archive")
remote := byID["pbs-backup:remote-archive"]
require.NotNil(t, remote.Verified)
require.False(t, *remote.Verified)
require.NotNil(t, remote.RepositoryRef)
require.Equal(t, "daily", remote.RepositoryRef.Name)
require.Equal(t, int64(4096), *remote.SizeBytes)
require.Equal(t, at, *remote.CompletedAt)
require.Contains(t, byID, "agent-backup")
}
type inventoryPointsProviderFunc func(context.Context, recovery.ListPointsOptions) ([]recovery.RecoveryPoint, int, error)
func (f inventoryPointsProviderFunc) ListPoints(ctx context.Context, opts recovery.ListPointsOptions) ([]recovery.RecoveryPoint, int, error) {
return f(ctx, opts)
}
func TestInvestigationBackupInventoryPagination(t *testing.T) {
all := make([]recovery.RecoveryPoint, 5010)
for i := range all {
all[i] = recovery.RecoveryPoint{ID: fmt.Sprintf("archive-%04d", i), Kind: recovery.KindBackup, SubjectResourceID: "vm-canonical"}
}
for _, tc := range []struct{ offset, limit, wantOffset, wantLimit, wantCount int }{
{499, 3, 499, 3, 3}, {5000, 2, 5000, 2, 2}, {5010, 3, 5010, 3, 0},
{-1, 0, 0, recovery.DefaultListPageLimit, recovery.DefaultListPageLimit},
{0, 900, 0, recovery.MaxListPageLimit, recovery.MaxListPageLimit},
} {
t.Run(fmt.Sprintf("offset-%d-limit-%d", tc.offset, tc.limit), func(t *testing.T) {
var queries []recovery.ListPointsOptions
executor := NewPulseToolExecutor(ExecutorConfig{RecoveryPointsProvider: inventoryPointsProviderFunc(func(ctx context.Context, opts recovery.ListPointsOptions) ([]recovery.RecoveryPoint, int, error) {
queries = append(queries, opts)
require.Equal(t, recovery.KindBackup, opts.Kind)
require.Equal(t, "vm-canonical", opts.SubjectResourceID)
return recovery.PaginatePoints(all, opts.Page, opts.Limit), len(all), nil
})})
result, err := executor.executeListBackups(context.Background(), map[string]interface{}{"resource_id": "vm-canonical", "offset": tc.offset, "limit": tc.limit})
require.NoError(t, err)
require.False(t, result.IsError, result.Content)
var response struct {
Points []recovery.RecoveryPoint `json:"points"`
Pagination PaginationInfo `json:"pagination"`
}
require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &response))
require.Len(t, response.Points, tc.wantCount)
require.Equal(t, PaginationInfo{Total: len(all), Offset: tc.wantOffset, Limit: tc.wantLimit}, response.Pagination)
if tc.wantCount > 0 {
require.Equal(t, all[tc.wantOffset:tc.wantOffset+tc.wantCount], response.Points)
}
require.LessOrEqual(t, len(queries), 2, "offset pagination must not scan earlier pages")
})
}
}
func TestInvestigationBackupInventoryDoesNotHideReadFailures(t *testing.T) {
for _, provider := range []RecoveryPointsProvider{nil, inventoryPointsProviderFunc(func(context.Context, recovery.ListPointsOptions) ([]recovery.RecoveryPoint, int, error) {
return nil, 0, context.Canceled
})} {
executor := NewPulseToolExecutor(ExecutorConfig{RecoveryPointsProvider: provider, BackupProvider: &stubBackupProvider{}})
result, err := executor.executeListBackups(context.Background(), nil)
require.NoError(t, err)
require.True(t, result.IsError, "unavailable inventory must not become an empty successful result")
}
}
func TestInvestigationReadsCanonicalBackupFailureWithoutAgent(t *testing.T) {
ctx := context.Background()
manager := recoverymanager.New(config.NewMultiTenantPersistence(t.TempDir()))
+36 -80
View File
@@ -19,7 +19,7 @@ func (e *PulseToolExecutor) registerStorageTools() {
e.registry.registerBuiltin(RegisteredTool{
Definition: Tool{
Name: agentcapabilities.PulseStorageToolName,
Description: `Query storage pools, backups, snapshots, Ceph, replication, RAID, and disk health. Use the "type" parameter to select what to query. backup_tasks returns collected backup attempt status, timestamps and failure error text without requiring host command access. Task success alone does not prove a restorable backup exists.`,
Description: `Query storage pools, backups, snapshots, Ceph, replication, RAID, and disk health. Use the "type" parameter to select what to query. backups returns collected backup artifact records from the canonical recovery inventory, including repository references and verification evidence when collected. backup_tasks returns collected backup attempt status, timestamps and failure error text without requiring host command access. Neither an artifact record nor task success alone proves a restorable backup exists.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
@@ -34,7 +34,7 @@ func (e *PulseToolExecutor) registerStorageTools() {
},
"resource_id": {
Type: "string",
Description: "Filter by resource ID (canonical subject ID for backup_tasks; VM/container ID for backups, snapshots, resource_disks)",
Description: "Filter by resource ID (canonical subject ID for backups and backup_tasks; VM/container ID for snapshots, resource_disks)",
},
"guest_id": {
Type: "string",
@@ -137,96 +137,52 @@ func (e *PulseToolExecutor) executeStorage(ctx context.Context, args map[string]
}
}
func (e *PulseToolExecutor) executeListBackups(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
func (e *PulseToolExecutor) executeListBackups(ctx context.Context, args map[string]interface{}) (CallToolResult, error) {
resourceID, _ := args["resource_id"].(string)
limit := intArg(args, "limit", 100)
offset := intArg(args, "offset", 0)
if e.backupProvider == nil {
return NewTextResult("Backup information not available."), nil
if limit <= 0 {
limit = recovery.DefaultListPageLimit
}
if limit > recovery.MaxListPageLimit {
limit = recovery.MaxListPageLimit
}
if offset < 0 {
offset = 0
}
if e.recoveryPointsProvider == nil {
return NewErrorResult(fmt.Errorf("backup inventory unavailable: recovery points provider not configured")), nil
}
backups := e.backupProvider.GetBackups()
pbsInstances := e.backupProvider.GetPBSInstances()
response := EmptyBackupsResponse()
response.Pagination = &PaginationInfo{Limit: limit, Offset: offset}
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
// PBS Backups
count := 0
for _, b := range backups.PBS {
if resourceID != "" && b.VMID != resourceID {
continue
// Translate the tool's global offset into the canonical store's pages.
// At most two pages are needed, even for offsets beyond the first 4,000
// records. Do not scan or paginate each provider's legacy lists separately.
const pageLimit = recovery.MaxListPageLimit
page, skip := offset/pageLimit+1, offset%pageLimit
for len(response.Points) < limit {
points, total, err := e.recoveryPointsProvider.ListPoints(ctx, recovery.ListPointsOptions{
Kind: recovery.KindBackup, SubjectResourceID: strings.TrimSpace(resourceID),
Page: page, Limit: pageLimit,
})
if err != nil {
return NewErrorResult(err), nil
}
if count < offset {
count++
continue
}
if len(response.PBS) >= limit {
response.Pagination.Total = total
if skip >= len(points) {
break
}
response.PBS = append(response.PBS, PBSBackupSummary{
VMID: b.VMID,
BackupType: b.BackupType,
BackupTime: b.BackupTime,
Instance: b.Instance,
Datastore: b.Datastore,
SizeGB: float64(b.Size) / (1024 * 1024 * 1024),
Verified: b.Verified,
Protected: b.Protected,
})
count++
}
// PVE Backups
count = 0
for _, b := range backups.PVE.StorageBackups {
if resourceID != "" && string(rune(b.VMID)) != resourceID {
continue
}
if count < offset {
count++
continue
}
if len(response.PVE) >= limit {
end := skip + min(limit-len(response.Points), len(points)-skip)
response.Points = append(response.Points, points[skip:end]...)
if end < len(points) || len(points) < pageLimit {
break
}
response.PVE = append(response.PVE, PVEBackupSummary{
VMID: b.VMID,
BackupTime: b.Time,
SizeGB: float64(b.Size) / (1024 * 1024 * 1024),
Storage: b.Storage,
})
count++
}
// PBS Servers
for _, pbs := range pbsInstances {
server := PBSServerSummary{
Name: pbs.Name,
Host: pbs.Host,
Status: pbs.Status,
}
for _, ds := range pbs.Datastores {
server.Datastores = append(server.Datastores, DatastoreSummary{
Name: ds.Name,
UsagePercent: ds.Usage,
FreeGB: float64(ds.Free) / (1024 * 1024 * 1024),
})
}
response.PBSServers = append(response.PBSServers, server)
}
// Recent tasks
for _, t := range backups.PVE.BackupTasks {
if len(response.RecentTasks) >= 20 {
break
}
response.RecentTasks = append(response.RecentTasks, BackupTaskSummary{
VMID: t.VMID,
Node: t.Node,
Status: t.Status,
StartTime: t.StartTime,
})
page++
skip = 0
}
return NewJSONResult(response.NormalizeCollections()), nil
+3 -97
View File
@@ -4,7 +4,6 @@ import (
"context"
"encoding/json"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
@@ -71,75 +70,8 @@ func (s *stubUpdatesProvider) IsUpdateActionsEnabled() bool {
return s.enabled
}
func TestExecuteListBackupsAndStorage(t *testing.T) {
func TestExecuteListStorage(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
executor.backupProvider = &stubBackupProvider{
backups: models.Backups{
PBS: []models.PBSBackup{
{
VMID: "100",
BackupType: "vm",
BackupTime: time.Unix(1000, 0),
Instance: "pbs1",
Datastore: "ds1",
Size: 1024 * 1024 * 1024,
Verified: true,
Protected: true,
},
},
PVE: models.PVEBackups{
StorageBackups: []models.StorageBackup{
{
VMID: 101,
Time: time.Unix(1100, 0),
Size: 2 * 1024 * 1024 * 1024,
Storage: "local",
},
},
BackupTasks: []models.BackupTask{
{
VMID: 101,
Node: "node1",
Status: "OK",
StartTime: time.Unix(1200, 0),
},
},
},
},
pbs: []models.PBSInstance{
{
Name: "pbs1",
Host: "10.0.0.1",
Status: "online",
Datastores: []models.PBSDatastore{
{
Name: "ds1",
Usage: 50.0,
Free: 1024 * 1024 * 1024,
},
},
},
},
}
result, _ := executor.executeListBackups(context.Background(), map[string]interface{}{})
var backupsResp BackupsResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &backupsResp); err != nil {
t.Fatalf("decode backups response: %v", err)
}
if len(backupsResp.PBS) != 1 || backupsResp.PBS[0].SizeGB != 1 {
t.Fatalf("unexpected PBS backups: %+v", backupsResp.PBS)
}
if len(backupsResp.PVE) != 1 || backupsResp.PVE[0].SizeGB != 2 {
t.Fatalf("unexpected PVE backups: %+v", backupsResp.PVE)
}
if len(backupsResp.PBSServers) != 1 || len(backupsResp.PBSServers[0].Datastores) != 1 {
t.Fatalf("unexpected PBS servers: %+v", backupsResp.PBSServers)
}
if len(backupsResp.RecentTasks) != 1 {
t.Fatalf("unexpected recent tasks: %+v", backupsResp.RecentTasks)
}
// Storage pools and Ceph data come from the unified resource provider.
storageUsedBytes := int64(1024 * 1024 * 1024)
storageTotalBytes := int64(4 * 1024 * 1024 * 1024)
@@ -192,7 +124,7 @@ func TestExecuteListBackupsAndStorage(t *testing.T) {
},
}
result, _ = executor.executeListStorage(context.Background(), map[string]interface{}{})
result, _ := executor.executeListStorage(context.Background(), map[string]interface{}{})
var storageResp StorageResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &storageResp); err != nil {
t.Fatalf("decode storage response: %v", err)
@@ -214,32 +146,6 @@ func TestStorageUsagePercentNotDoubled(t *testing.T) {
// The bug was multiplying by 100 again, turning 25% into 2500%.
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
// PBS datastore path
executor.backupProvider = &stubBackupProvider{
pbs: []models.PBSInstance{
{
Name: "pbs1",
Host: "10.0.0.1",
Status: "online",
Datastores: []models.PBSDatastore{
{Name: "ds1", Usage: 12.3, Free: 1024 * 1024 * 1024},
},
},
},
}
result, _ := executor.executeListBackups(context.Background(), map[string]interface{}{})
var backupsResp BackupsResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &backupsResp); err != nil {
t.Fatalf("decode backups response: %v", err)
}
if len(backupsResp.PBSServers) != 1 || len(backupsResp.PBSServers[0].Datastores) != 1 {
t.Fatalf("unexpected PBS servers: %+v", backupsResp.PBSServers)
}
got := backupsResp.PBSServers[0].Datastores[0].UsagePercent
if got != 12.3 {
t.Errorf("PBS datastore UsagePercent = %v, want 12.3 (double-multiplied = %v)", got, 12.3*100)
}
// Storage pool path
storageUsedBytes := int64(1024 * 1024 * 1024)
storageTotalBytes := int64(4 * 1024 * 1024 * 1024)
@@ -346,7 +252,7 @@ func TestStorageResponsesUseCanonicalEmptyCollections(t *testing.T) {
if err := json.Unmarshal(backupsPayload, &backupsMap); err != nil {
t.Fatalf("decode empty backups response: %v", err)
}
for _, key := range []string{"pbs", "pve", "pbs_servers", "recent_tasks"} {
for _, key := range []string{"points"} {
values, ok := backupsMap[key].([]any)
if !ok || len(values) != 0 {
t.Fatalf("expected %s to serialize as an empty array, got %T (%v)", key, backupsMap[key], backupsMap[key])
@@ -91,6 +91,58 @@ func TestInvestigationBackupTasksUseCommunityRecoverySource(t *testing.T) {
}
}
func TestInvestigationBackupInventoryMatchesCommunityPoints(t *testing.T) {
testutil.SetMockMode(t, true)
manager := newRecoveryReadManager(config.NewMultiTenantPersistence(t.TempDir()))
var selected recovery.RecoveryPoint
for _, point := range mock.CurrentFixtureGraph().RecoveryPoints() {
if point.Kind == recovery.KindBackup && point.SubjectResourceID != "" {
selected = point
break
}
}
require.NotEmpty(t, selected.ID)
store, err := manager.StoreForOrg("default")
require.NoError(t, err)
t.Cleanup(func() { _ = store.Close() })
require.NoError(t, store.UpsertPoints(context.Background(), []recovery.RecoveryPoint{selected}))
executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{RecoveryPointsProvider: tools.NewRecoveryPointsToolAdapter(manager, "default")})
for _, demo := range []bool{true, false, true} {
require.NoError(t, mock.SetEnabled(demo))
for _, resourceID := range []string{selected.SubjectResourceID, "vm-missing"} {
recorder := httptest.NewRecorder()
NewRecoveryHandlers(manager).HandleListPoints(recorder, httptest.NewRequest(http.MethodGet,
"/api/recovery/points?kind=backup&limit=500&itemResourceId="+url.QueryEscape(resourceID), nil))
require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
var apiResponse struct {
Data []recovery.RecoveryPoint `json:"data"`
Meta struct {
Total int `json:"total"`
} `json:"meta"`
}
require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &apiResponse))
result, err := executor.ExecuteTool(context.Background(), "pulse_storage", map[string]interface{}{
"type": "backups", "resource_id": resourceID, "limit": 500,
})
require.NoError(t, err)
require.False(t, result.IsError, result.Content)
var toolResponse struct {
Points []recovery.RecoveryPoint `json:"points"`
Pagination tools.PaginationInfo `json:"pagination"`
}
require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &toolResponse))
require.Equal(t, apiResponse.Meta.Total, toolResponse.Pagination.Total)
require.Equal(t, apiResponse.Data, toolResponse.Points, "inventory identity, repository, verification and provenance must match HTTP")
if resourceID == selected.SubjectResourceID {
require.NotEmpty(t, toolResponse.Points)
} else {
require.NotNil(t, toolResponse.Points)
require.Empty(t, toolResponse.Points)
}
}
}
}
type investigationHistoryResources struct{ resource unifiedresources.Resource }
func (p investigationHistoryResources) GetByType(kind unifiedresources.ResourceType) []unifiedresources.Resource {