wmi: add generic wmi collector (#2530)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>

internal/mi/mi_test.go

@@ -873,3 +873,119 @@ func Test_MI_QueryUnmarshal_Uint16Array_WrongType(t *testing.T) {
 	err = application.Close()
 	require.NoError(t, err)
 }
+
+func Test_MI_QueryFunc(t *testing.T) {
+	application, err := mi.ApplicationInitialize()
+	require.NoError(t, err)
+
+	t.Cleanup(func() { require.NoError(t, application.Close()) })
+
+	session, err := application.NewSession(nil)
+	require.NoError(t, err)
+
+	t.Cleanup(func() { require.NoError(t, session.Close()) })
+
+	query, err := mi.NewQuery("SELECT LocalDateTime, NumberOfProcesses, Caption FROM Win32_OperatingSystem")
+	require.NoError(t, err)
+
+	var calls int
+
+	err = session.QueryFunc(mi.NamespaceRootCIMv2, query, 5*time.Second, func(instance *mi.Instance) error {
+		calls++
+
+		// LocalDateTime is a DATETIME timestamp carrying the local UTC offset;
+		// decoding it must land close to the current wall clock.
+		element, err := instance.GetElement("LocalDateTime")
+		require.NoError(t, err)
+		require.False(t, element.IsNull())
+
+		localDateTime, err := element.Float64()
+		require.NoError(t, err)
+		require.InDelta(t, float64(time.Now().Unix()), localDateTime, 60)
+
+		element, err = instance.GetElement("NumberOfProcesses")
+		require.NoError(t, err)
+
+		processes, err := element.Float64()
+		require.NoError(t, err)
+		require.Positive(t, processes)
+
+		element, err = instance.GetElement("Caption")
+		require.NoError(t, err)
+
+		caption, err := element.String()
+		require.NoError(t, err)
+		require.Contains(t, caption, "Windows")
+
+		return nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, 1, calls)
+
+	// An error returned by the callback aborts the query and is passed through.
+	errStop := errors.New("stop")
+
+	query, err = mi.NewQuery("SELECT Name FROM Win32_Process")
+	require.NoError(t, err)
+
+	calls = 0
+	err = session.QueryFunc(mi.NamespaceRootCIMv2, query, 5*time.Second, func(*mi.Instance) error {
+		calls++
+
+		return errStop
+	})
+	require.ErrorIs(t, err, errStop)
+	require.Equal(t, 1, calls)
+
+	// The cancelled query must leave the session usable.
+	calls = 0
+	err = session.QueryFunc(mi.NamespaceRootCIMv2, query, 5*time.Second, func(*mi.Instance) error {
+		calls++
+
+		return nil
+	})
+	require.NoError(t, err)
+	require.Greater(t, calls, 1)
+
+	// Invalid classes surface as an error rather than an empty result.
+	query, err = mi.NewQuery("SELECT Name FROM Win32_DoesNotExist")
+	require.NoError(t, err)
+
+	err = session.QueryFunc(mi.NamespaceRootCIMv2, query, 5*time.Second, func(*mi.Instance) error {
+		return nil
+	})
+	require.Error(t, err)
+}
+
+func Test_MI_Operation_CloseEarly(t *testing.T) {
+	application, err := mi.ApplicationInitialize()
+	require.NoError(t, err)
+
+	t.Cleanup(func() { require.NoError(t, application.Close()) })
+
+	session, err := application.NewSession(nil)
+	require.NoError(t, err)
+
+	t.Cleanup(func() { require.NoError(t, session.Close()) })
+
+	// Closing an operation with pending results cancels it instead of reading
+	// them, and must leave the session usable.
+	for range 3 {
+		operation, err := session.QueryInstances(mi.OperationFlagsStandardRTTI, nil, mi.NamespaceRootCIMv2, mi.QueryDialectWQL, "SELECT Name FROM Win32_Process")
+		require.NoError(t, err)
+
+		instance, moreResults, err := operation.GetInstance()
+		require.NoError(t, err)
+		require.NotNil(t, instance)
+		require.True(t, moreResults)
+
+		require.NoError(t, operation.Close())
+	}
+
+	var processes []win32Process
+
+	query, err := mi.NewQuery("SELECT Name FROM Win32_Process WHERE Handle = 0")
+	require.NoError(t, err)
+	require.NoError(t, session.Query(&processes, mi.NamespaceRootCIMv2, query, 5*time.Second))
+	require.Equal(t, []win32Process{{Name: "System Idle Process"}}, processes)
+}