CLI: adding more validation to workflow

packages/cli/src/execute/handler.ts

@@ -16,6 +16,7 @@ import assertPath from '../util/assert-path';
 import { clearCache } from '../util/cache';
 import fuzzyMatchStep from '../util/fuzzy-match-step';
 import abort from '../util/abort';
+import validatePlan from '../util/validate-plan';
 
 const matchStep = (
   plan: ExecutionPlan,
@@ -48,6 +49,7 @@ const executeHandler = async (options: ExecuteOptions, logger: Logger) => {
   await validateAdaptors(options, logger);
 
   let plan = await loadPlan(options, logger);
+  validatePlan(plan, logger);
 
   if (options.cacheSteps) {
     await clearCache(plan, options, logger);

packages/cli/src/util/validate-plan.ts

@@ -0,0 +1,50 @@
+import { ExecutionPlan, Step, WorkflowOptions } from "@openfn/lexicon";
+import { Logger } from "@openfn/logger";
+
+const assertWorkflowStructure = (plan: ExecutionPlan, logger: Logger) => {
+    const { workflow, options } = plan;
+
+    if (!workflow || typeof workflow !== 'object') {
+        throw new Error(`Missing or invalid "workflow" key in execution plan`);
+    }
+
+    if (!Array.isArray(workflow.steps)) {
+        throw new Error('The workflow.steps key must be an array');
+    }
+
+    if (workflow.steps.length === 0) {
+        logger.warn('The workflow.steps array is empty');
+    }
+
+    workflow.steps.forEach((step, index) => {
+        assertStepStructure(step, index);
+    });
+
+    assertOptionsStructure(options, logger);
+};
+
+const assertStepStructure = (step: Step, index: number) => {
+    const allowedKeys = ['id', 'name', 'next', 'previous', 'adaptor', 'expression', 'state', 'configuration', 'linker'];
+
+    for (const key in step) {
+        if (!allowedKeys.includes(key)) {
+            throw new Error(`Invalid key "${key}" in step ${step.id || index}`);
+        }
+    }
+
+    if ('adaptor' in step && !('expression' in step)) {
+        throw new Error(`Step ${step.id ?? index} with an adaptor must also have an expression`);
+    }
+};
+
+const assertOptionsStructure = (options: WorkflowOptions = {}, logger: Logger) => {
+    const allowedKeys = ['timeout', 'stepTimeout', 'start', 'end', 'sanitize'];
+
+    for (const key in options) {
+        if (!allowedKeys.includes(key)) {
+            logger.warn(`Unrecognized option "${key}" in options object`);
+        }
+    }
+};
+
+export default assertWorkflowStructure;

packages/cli/test/util/validate-plan.test.ts

@@ -0,0 +1,113 @@
+import test from 'ava';
+import { createMockLogger } from '@openfn/logger';
+import type { ExecutionPlan } from '@openfn/lexicon';
+import validate from '../../src/util/validate-plan';
+
+const logger = createMockLogger('', { level: 'debug' });
+
+test.afterEach(() => {
+    logger._reset();
+})
+
+test('throws for missing workflow', (t) => {
+    const plan = {
+        options: {
+            start: 'a',
+        }
+    } as ExecutionPlan;
+
+    t.throws(() => validate(plan, logger), {
+        message: `Missing or invalid "workflow" key in execution plan`,
+    });
+});
+
+test('throws for steps not an array', (t) => {
+
+    const plan = {
+        options: {
+            start: 'a',
+        },
+        workflow: {
+            steps: {
+                id: 'a'
+            }
+        },
+    } as unknown as ExecutionPlan;
+
+    t.throws(() => validate(plan, logger), {
+        message: 'The workflow.steps key must be an array',
+    });
+});
+
+test('throws for a step with an adaptor but no expression', (t) => {
+    const plan = {
+        options: {
+            start: 'a',
+        },
+        workflow: {
+            steps: [
+                {
+                    id: 'a',
+                    adaptor: 'z'
+                }
+            ],
+        },
+    } as unknown as ExecutionPlan;
+
+    t.throws(() => validate(plan, logger), {
+        message: 'Step a with an adaptor must also have an expression',
+    });
+});
+
+test('throws for unknown key in a step', (t) => {
+    const plan = {
+        options: {
+            start: 'a',
+        },
+        workflow: {
+            steps: [
+                {
+                    id: 'a',
+                    key: 'z'
+                }
+            ],
+        },
+    }as unknown as ExecutionPlan;
+
+    t.throws(() => validate(plan, logger), {
+        message: 'Invalid key "key" in step a',
+    });
+});
+
+test.serial('should warn if no steps are defined', (t) => {
+    const plan: ExecutionPlan = {
+        options: {
+            start: 'a',
+        },
+        workflow: {
+            steps: [],
+        },
+    };
+    validate(plan, logger);
+    const { message, level } = logger._parse(logger._history[0]);
+    t.is(level, 'warn');
+    t.regex(message as string, /The workflow.steps array is empty/);
+})
+
+test.serial('should warn if unknown key is passed in options', (t) => {
+    const plan = {
+        options: {
+            start: 'a',
+            key: 'z',
+        },
+        workflow: {
+            steps: [{
+                id: 'a',
+            }],
+        },
+    } as unknown as ExecutionPlan;
+    validate(plan, logger);
+    const { message, level } = logger._parse(logger._history[0]);
+    t.is(level, 'warn');
+    t.regex(message as string, /Unrecognized option "key" in options object/);
+})

packages/runtime/src/execute/plan.ts

@@ -91,4 +91,4 @@ const executePlan = async (
   return Object.values(leaves)[0];
 };
 
-export default executePlan;
+export default executePlan;

packages/runtime/src/util/validate-plan.ts

@@ -116,4 +116,4 @@ const assertSingletonDependencies = (model: Model) => {
       throw new ValidationError(`Multiple dependencies detected for: ${id}`);
     }
   }
-};
+};

packages/runtime/test/runtime.test.ts

@@ -528,7 +528,9 @@ test('stuff written to state before an error is preserved', async (t) => {
       steps: [
         {
           id: 'a',
-          data: { x: 0 },
+          state: {
+            data: { x: 0 }
+          },
           expression:
             'export default [(s) => { s.x = 1; throw new Error("test") }]',
         },