test(k8s): add RWX and RWO affinity coverage

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>

Author: nikola-jokic

packages/k8s/tests/rwo-affinity-test.ts

@@ -0,0 +1,143 @@
+import * as fs from 'fs'
+import { cleanupJob } from '../src/hooks'
+import { prepareJob } from '../src/hooks/prepare-job'
+import { TestHelper } from './test-setup'
+import { getPodByName } from '../src/k8s'
+import { ENV_USE_KUBE_SCHEDULER } from '../src/k8s/utils'
+
+jest.useRealTimers()
+
+let testHelper: TestHelper
+let prepareJobData: any
+let prepareJobOutputFilePath: string
+
+describe('RWO Affinity Behavior (Scheduler Mode)', () => {
+  beforeEach(async () => {
+    testHelper = new TestHelper()
+    await testHelper.initialize()
+    prepareJobData = testHelper.getPrepareJobDefinition()
+    prepareJobOutputFilePath = testHelper.createFile('prepare-job-output.json')
+  })
+
+  afterEach(async () => {
+    await cleanupJob()
+    await testHelper.cleanup()
+    delete process.env[ENV_USE_KUBE_SCHEDULER]
+  })
+
+  it('should add nodeAffinity with hostname selector when scheduler mode is enabled', async () => {
+    process.env[ENV_USE_KUBE_SCHEDULER] = 'true'
+
+    await prepareJob(prepareJobData.args, prepareJobOutputFilePath)
+
+    const content = JSON.parse(
+      fs.readFileSync(prepareJobOutputFilePath).toString()
+    )
+
+    const pod = await getPodByName(content.state.jobPod)
+
+    expect(pod.spec?.affinity).toBeDefined()
+    expect(pod.spec?.affinity?.nodeAffinity).toBeDefined()
+
+    const nodeAffinity = pod.spec?.affinity?.nodeAffinity
+    expect(
+      nodeAffinity?.requiredDuringSchedulingIgnoredDuringExecution
+    ).toBeDefined()
+
+    const nodeSelectorTerms =
+      nodeAffinity?.requiredDuringSchedulingIgnoredDuringExecution
+        ?.nodeSelectorTerms
+
+    expect(nodeSelectorTerms).toBeDefined()
+    expect(nodeSelectorTerms?.length).toBeGreaterThan(0)
+
+    const matchExpressions = nodeSelectorTerms?.[0].matchExpressions
+    expect(matchExpressions).toBeDefined()
+    expect(matchExpressions?.length).toBeGreaterThan(0)
+
+    const hostnameExpression = matchExpressions?.[0]
+    expect(hostnameExpression?.key).toBe('kubernetes.io/hostname')
+    expect(hostnameExpression?.operator).toBe('In')
+
+    expect(hostnameExpression?.values).toBeDefined()
+    expect(hostnameExpression?.values?.length).toBeGreaterThan(0)
+    expect(hostnameExpression?.values?.[0]).toBeTruthy()
+  })
+
+  it('should NOT add nodeAffinity when scheduler mode is disabled', async () => {
+    process.env[ENV_USE_KUBE_SCHEDULER] = 'false'
+
+    await prepareJob(prepareJobData.args, prepareJobOutputFilePath)
+
+    const content = JSON.parse(
+      fs.readFileSync(prepareJobOutputFilePath).toString()
+    )
+
+    const pod = await getPodByName(content.state.jobPod)
+
+    if (pod.spec?.affinity) {
+      expect(pod.spec.affinity.nodeAffinity).toBeUndefined()
+    }
+
+    expect(pod.spec?.nodeName).toBeDefined()
+  })
+
+  it('should fail assertion if affinity block is missing when scheduler mode is enabled', async () => {
+    process.env[ENV_USE_KUBE_SCHEDULER] = 'true'
+
+    await prepareJob(prepareJobData.args, prepareJobOutputFilePath)
+
+    const content = JSON.parse(
+      fs.readFileSync(prepareJobOutputFilePath).toString()
+    )
+
+    const pod = await getPodByName(content.state.jobPod)
+
+    expect(pod.spec?.affinity).toBeDefined()
+    expect(pod.spec?.affinity?.nodeAffinity).toBeDefined()
+    expect(
+      pod.spec?.affinity?.nodeAffinity
+        ?.requiredDuringSchedulingIgnoredDuringExecution
+    ).toBeDefined()
+
+    const nodeSelectorTerms =
+      pod.spec?.affinity?.nodeAffinity
+        ?.requiredDuringSchedulingIgnoredDuringExecution?.nodeSelectorTerms
+
+    expect(nodeSelectorTerms?.[0]?.matchExpressions?.[0]?.key).toBe(
+      'kubernetes.io/hostname'
+    )
+    expect(nodeSelectorTerms?.[0]?.matchExpressions?.[0]?.operator).toBe('In')
+    expect(
+      nodeSelectorTerms?.[0]?.matchExpressions?.[0]?.values?.length
+    ).toBeGreaterThan(0)
+  })
+
+  it('should use correct node name from runner pod in affinity values', async () => {
+    process.env[ENV_USE_KUBE_SCHEDULER] = 'true'
+
+    const runnerPodName = process.env.ACTIONS_RUNNER_POD_NAME
+
+    await prepareJob(prepareJobData.args, prepareJobOutputFilePath)
+
+    const content = JSON.parse(
+      fs.readFileSync(prepareJobOutputFilePath).toString()
+    )
+
+    const jobPod = await getPodByName(content.state.jobPod)
+
+    const runnerPod = await getPodByName(runnerPodName!)
+
+    const affinityValues =
+      jobPod.spec?.affinity?.nodeAffinity
+        ?.requiredDuringSchedulingIgnoredDuringExecution?.nodeSelectorTerms?.[0]
+        ?.matchExpressions?.[0]?.values
+
+    expect(affinityValues).toBeDefined()
+    expect(affinityValues?.length).toBeGreaterThan(0)
+
+    if (runnerPod.spec?.nodeName) {
+      expect(affinityValues).toContain(runnerPod.spec.nodeName)
+    }
+  })
+})

packages/k8s/tests/rwx-contract-demo-test.ts

@@ -0,0 +1,28 @@
+import {
+  isRWXTestEnabled,
+  getRWXStorageClass,
+  RWX_SKIP_MESSAGE
+} from './test-setup'
+
+describe('RWX Test Contract Demo', () => {
+  const describeOrSkip = isRWXTestEnabled() ? describe : describe.skip
+
+  describeOrSkip('RWX volume tests', () => {
+    it('should use RWX storage class when enabled', () => {
+      const storageClass = getRWXStorageClass()
+      expect(storageClass).toBeDefined()
+      expect(typeof storageClass).toBe('string')
+    })
+
+    it('should verify both env vars are required', () => {
+      expect(process.env.ACTIONS_RUNNER_K8S_TEST_ENABLE_RWX).toBe('true')
+      expect(
+        process.env.ACTIONS_RUNNER_K8S_TEST_RWX_STORAGE_CLASS
+      ).toBeDefined()
+    })
+  })
+
+  if (!isRWXTestEnabled()) {
+    it(RWX_SKIP_MESSAGE, () => {})
+  }
+})

packages/k8s/tests/rwx-volume-test.ts

@@ -0,0 +1,119 @@
+import * as k8s from '@kubernetes/client-node'
+import * as fs from 'fs'
+import { cleanupJob, prepareJob, runScriptStep } from '../src/hooks'
+import {
+  TestHelper,
+  isRWXTestEnabled,
+  getRWXStorageClass,
+  RWX_SKIP_MESSAGE
+} from './test-setup'
+import { RunScriptStepArgs } from 'hooklib'
+
+jest.useRealTimers()
+
+const kc = new k8s.KubeConfig()
+kc.loadFromDefault()
+const k8sApi = kc.makeApiClient(k8s.CoreV1Api)
+
+describe('RWX Volume Tests', () => {
+  const describeOrSkip = isRWXTestEnabled() ? describe : describe.skip
+
+  describeOrSkip('RWX volume integration', () => {
+    let testHelper: TestHelper
+    let rwxPvcName: string
+    let prepareJobData: any
+    let prepareJobOutputFilePath: string
+
+    beforeEach(async () => {
+      testHelper = new TestHelper()
+      await testHelper.initialize()
+
+      const podName = process.env.ACTIONS_RUNNER_POD_NAME
+      rwxPvcName = `${podName}-work-rwx`
+
+      const volumeClaim: k8s.V1PersistentVolumeClaim = {
+        metadata: {
+          name: rwxPvcName
+        },
+        spec: {
+          accessModes: ['ReadWriteMany'],
+          volumeMode: 'Filesystem',
+          storageClassName: getRWXStorageClass(),
+          resources: {
+            requests: {
+              storage: '1Gi'
+            }
+          }
+        }
+      }
+
+      await k8sApi.createNamespacedPersistentVolumeClaim({
+        namespace: 'default',
+        body: volumeClaim
+      })
+
+      process.env.ACTIONS_RUNNER_CLAIM_NAME = rwxPvcName
+
+      prepareJobData = testHelper.getPrepareJobDefinition()
+      prepareJobOutputFilePath = testHelper.createFile(
+        'prepare-job-output.json'
+      )
+    })
+
+    afterAll(async () => {
+      if (rwxPvcName) {
+        try {
+          await k8sApi.deleteNamespacedPersistentVolumeClaim({
+            name: rwxPvcName,
+            namespace: 'default'
+          })
+        } catch {
+          // Ignore cleanup errors - PVC may not exist
+        }
+      }
+    })
+
+    afterEach(async () => {
+      await testHelper.cleanup()
+    })
+
+    it('should successfully run hook flow with RWX volume', async () => {
+      await expect(
+        prepareJob(prepareJobData.args, prepareJobOutputFilePath)
+      ).resolves.not.toThrow()
+
+      const prepareJobOutputJson = fs.readFileSync(prepareJobOutputFilePath)
+      const prepareJobOutputData = JSON.parse(prepareJobOutputJson.toString())
+
+      const scriptStepData = testHelper.getRunScriptStepDefinition()
+
+      await expect(
+        runScriptStep(
+          scriptStepData.args as RunScriptStepArgs,
+          prepareJobOutputData.state
+        )
+      ).resolves.not.toThrow()
+
+      await expect(cleanupJob()).resolves.not.toThrow()
+    })
+
+    it('should verify RWX PVC was created with correct access mode', async () => {
+      const pvc = await k8sApi.readNamespacedPersistentVolumeClaim({
+        name: rwxPvcName,
+        namespace: 'default'
+      })
+
+      expect(pvc.spec?.accessModes).toContain('ReadWriteMany')
+      expect(pvc.spec?.storageClassName).toBe(getRWXStorageClass())
+      expect(pvc.spec?.volumeMode).toBe('Filesystem')
+    })
+
+    it('should verify RWX claim name is set correctly', () => {
+      expect(process.env.ACTIONS_RUNNER_CLAIM_NAME).toBe(rwxPvcName)
+    })
+  })
+
+  if (!isRWXTestEnabled()) {
+    it(RWX_SKIP_MESSAGE, () => {})
+  }
+})

packages/k8s/tests/test-setup.ts

@@ -243,3 +243,65 @@ export class TestHelper {
     return runContainerStep
   }
 }
+
+/**
+ * RWX Test Contract:
+ *
+ * Tests requiring ReadWriteMany (RWX) volumes MUST be gated by TWO environment variables:
+ * 1. ACTIONS_RUNNER_K8S_TEST_ENABLE_RWX=true (explicit opt-in)
+ * 2. ACTIONS_RUNNER_K8S_TEST_RWX_STORAGE_CLASS=<name> (storage class that supports RWX)
+ *
+ * If either variable is missing or ACTIONS_RUNNER_K8S_TEST_ENABLE_RWX is not "true",
+ * the test MUST be skipped with the exact message defined in this contract.
+ *
+ * This contract ensures:
+ * - RWX tests do not fail on clusters without RWX provisioners
+ * - Test requirements are explicit and documented
+ * - RWO affinity tests remain independent and always runnable
+ * - Skip behavior is deterministic (no dynamic cluster probing)
+ *
+ * Usage example:
+ * ```typescript
+ * import { isRWXTestEnabled, getRWXStorageClass, RWX_SKIP_MESSAGE } from './test-setup'
+ *
+ * describe('RWX Test Suite', () => {
+ *   const describeOrSkip = isRWXTestEnabled() ? describe : describe.skip
+ *
+ *   describeOrSkip('RWX volume tests', () => {
+ *     it('should test RWX functionality', async () => {
+ *       const storageClass = getRWXStorageClass()
+ *       // ... test code using storageClass
+ *     })
+ *   })
+ *
+ *   if (!isRWXTestEnabled()) {
+ *     it(RWX_SKIP_MESSAGE, () => {})
+ *   }
+ * })
+ * ```
+ */
+
+/**
+ * Checks if RWX tests should run based on environment variables.
+ * @returns true if both ACTIONS_RUNNER_K8S_TEST_ENABLE_RWX=true and ACTIONS_RUNNER_K8S_TEST_RWX_STORAGE_CLASS are set
+ */
+export function isRWXTestEnabled(): boolean {
+  const enabled = process.env.ACTIONS_RUNNER_K8S_TEST_ENABLE_RWX === 'true'
+  const storageClass = process.env.ACTIONS_RUNNER_K8S_TEST_RWX_STORAGE_CLASS
+  return enabled && !!storageClass
+}
+
+/**
+ * Gets the RWX storage class name from environment variable.
+ * @returns The storage class name, or undefined if not set
+ */
+export function getRWXStorageClass(): string | undefined {
+  return process.env.ACTIONS_RUNNER_K8S_TEST_RWX_STORAGE_CLASS
+}
+
+/**
+ * Skip message constant - DO NOT MODIFY
+ * This exact message must be used when skipping RWX tests
+ */
+export const RWX_SKIP_MESSAGE =
+  'RWX tests skipped: set ACTIONS_RUNNER_K8S_TEST_ENABLE_RWX=true and ACTIONS_RUNNER_K8S_TEST_RWX_STORAGE_CLASS'