Merge pull request #1 from MediaMath/json_schema

Use JSON-Schema to validate task.

Author: javibravo

.gitignore

@@ -19,5 +19,3 @@ sample/
 # IntelliJ IDEs
 .idea/
 
-# LambdaCron config
-config/

README.md

@@ -227,30 +227,44 @@ Parameters:
 * **--directory (-d)**: Path to directory that contains tasks definitions (string)
 * **--aws-profile (-a)**: AWS profile to use from aws-cli (string) (optional)
 
+#### validate
+
+Validate a tasks checking if they match with the schema. It can validate a task
+from a file or a set of tasks in a directory.
+
+Parameters:
+
+* **--task-file (-t)**: File that contains a task definition.
+* **--task-directory (-d)**: Directory with a set of files with taqsks definitions.
+
 
 ## Tasks
 
-Tasks are defined in YAML files, each task in an independent file. The keys that
-must contains every task are:
+Tasks are defined in YAML files, each task in an independent file. Task must follow
+the json schema provided in this repo: [schema](./schema.json).
+
+All tasks must contains following keys and values:
 
 * **name**: task name
 * **expression**: crontab expression
-* **type**: task type (**queue** | **lambda** | **http**)
 * **task**: task definition (customized for each type of tasks)
 
+For each kind of task there a set of specific keys a values to set. Described bellow.
+
 ### Queue task
 
 It sends a message to a AWS SQS queue.
 The task definition must contains following keys:
 
+* **type**: *queue*
 * **QueueName**: Name of the queue (string)
 * **MessageBody**: Message to be sent (YAML/JSON)
 
 ``` yaml
 name: 'Send performance report every morning'
 expression: '0 9 * * *'
-type: 'queue'
 task:
+  type: 'queue'
   QueueName: 'my-scheduler-queue'
   MessageBody:
     name: 'Performance report'
@@ -268,14 +282,15 @@ All parameters of the function will be supported soon.
 It invokes an AWS lambda functions.
 The task definition must contains following keys
 
+* **type**: *lambda*
 * **FunctionName**: Name of the lambda function to invoke (string)
 * **InvokeArgs**: arguments to send (YAML/JSON)
 
 ``` yaml
 name: 'Run ETL process every hour'
 expression: '0 * * * *'
-type: 'lambda'
 task:
+  type: 'lambda'
   FunctionName: 'run-etl-process-prod'
   InvokeArgs:
     source: 's3://my-data-source/performance'
@@ -289,14 +304,15 @@ Function is invoked using [boto3 Lambda.Client.invoke_async](http://boto3.readth
 It send and HTTP request (GET or POST).
 The task definition must contains following keys:
 
+* **type**: *http*
 * **method**: http method (get | post)
 * **request**: YAML with parameters to send for the selected method using [Requests](http://docs.python-requests.org/en/master/)
 
 ``` yaml
 name: 'helth check every hour'
 expression: '0 * * * *'
-type: 'http'
 task:
+  type: 'http'
   method: 'get'
   request:
     url: 'http://helthcheck.my-domain.com'
@@ -361,9 +377,10 @@ Create your first environment (called 'test') with default settings:
 $ bin/lambda-cron create --environment=test --create-bucket
 ```
 
-If you want to set some custom settings create the setting file:
+If you want to set some custom settings create the setting file in the home
+directory of the user is running the tool.
 
-* config/cli.yml
+* ~/lambdacron.yml
 
 For help:
 

cli/config_cli.py

@@ -10,7 +10,11 @@ def get_project_root_directory():
 
 
 def get_cli_config_file_path():
-    return os.path.join(get_project_root_directory(), 'config/cli.yml')
+    return os.path.abspath('~/.lambdacron.yml')
+
+
+def get_jsonschema_file_path():
+    return os.path.join(get_project_root_directory(), 'schema.json')
 
 
 def load_config():

cli/lambda_cron_cli.py

@@ -11,6 +11,8 @@
 from config_cli import ConfigCli
 import config_cli
 import yaml
+import json
+import jsonschema
 
 
 def check_arg(args=None):
@@ -48,6 +50,10 @@ def check_arg(args=None):
     deploy_command.add_argument('-d', '--directory', required=True)
     deploy_command.add_argument('-a', '--aws-profile', default=None, dest='aws_profile')
 
+    deploy_command = commands_parser.add_parser('validate')
+    deploy_command.add_argument('-t', '--task-file', required=False, dest='task_file')
+    deploy_command.add_argument('-d', '--task-directory', required=False, dest='task_directory')
+
     return parser.parse_args(args)
 
 
@@ -74,7 +80,8 @@ class LambdaCronCLI:
     def __init__(self, cli_instructions):
         self.cli = cli_instructions
         self.timestamp = int(round(time.time() * 1000))
-        self.config = ConfigCli(self.cli.environment)
+        if 'environment' in self.cli:
+            self.config = ConfigCli(self.cli.environment)
 
     def get_tmp_directory(self):
         return '/tmp/LambdaCron-{environment}'.format(environment=self.cli.environment)
@@ -306,6 +313,37 @@ def upload_tasks(self):
         ]
         self.exec_aws_command(delete_stack_command)
 
+    def validate_task(self, schema, task_file_name):
+        try:
+            with open(task_file_name, 'r') as task_file:
+                task = yaml.load(task_file)
+            jsonschema.validate(task, schema)
+        except Exception, ex:
+            raise ex
+
+    def validate(self):
+        try:
+            with open(config_cli.get_jsonschema_file_path(), 'r') as schema_file:
+                schema = json.load(schema_file)
+
+            if self.cli.task_file:
+                self.validate_task(schema, self.cli.task_file)
+
+            if self.cli.task_directory:
+                all_yml_files = [os.path.join(dirpath, f)
+                                 for dirpath, dirnames, files in os.walk(self.cli.task_directory)
+                                 for f in files if f.endswith('.yml')]
+                print all_yml_files
+                for file_name in all_yml_files:
+                    print file_name
+                    self.validate_task(schema, file_name)
+        except jsonschema.exceptions.ValidationError, ex:
+            print("Validation failed! Validation error in task: {}".format(ex.message))
+            sys.exit(1)
+        except Exception, ex:
+            raise ex
+        print 'Validation success!'
+
     def run(self):
         command_method = getattr(self, self.cli.command.replace('-', '_'))
         command_method()

lambda_cron/lib/task_runner.py

@@ -14,7 +14,7 @@ def __init__(self, cron_checker):
 
     def run(self, task):
         if self.cron_checker.should_run(task['expression']):
-            command_method = getattr(self, "get_{}_task_runner".format(task['type'].lower()))
+            command_method = getattr(self, "get_{}_task_runner".format(task['task']['type'].lower()))
             response = command_method(task['task']).run()
             logger.info('Task executed!')
             logger.info(response)

requirements.txt

@@ -1,4 +1,5 @@
 croniter==0.3.12
 python-dateutil==2.5.3
 PyYAML==3.12
-requests==2.12.3
+requests==2.12.3
+jsonschema==2.5.1

schema.json

@@ -0,0 +1,43 @@
+{
+  "$schema": "http://json-schema.org/schema",
+  "title": "LambdaCron v0.0.1",
+  "description": "JSON schema for LambdaCron tasks",
+  "properties": {
+    "name": {
+      "type": "string"
+    },
+    "expression": {
+      "type": "string",
+      "pattern": "(\\*|[0-5]?[0-9]|\\*\/[0-9]+)\\s+(\\*|1?[0-9]|2[0-3]|\\*\/[0-9]+)\\s+(\\*|[1-2]?[0-9]|3[0-1]|\\*\/[0-9]+)\\s+(\\*|[0-9]|1[0-2]|\\*\/[0-9]+|jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)\\s+(\\*\/[0-9]+|\\*|[0-7]|sun|mon|tue|wed|thu|fri|sat)\\s*(\\*\/[0-9]+|\\*|[0-9]+)?"
+    },
+    "task": {
+      "oneOf" : [
+        {
+          "properties": {
+            "type": { "type": "string", "enum": ["queue"] },
+            "QueueName": { "type": "string" },
+            "MessageBody": { "type": "object" }
+          },
+          "required": ["type", "QueueName", "MessageBody"]
+        },
+        {
+          "properties": {
+            "type": { "type": "string", "enum": ["lambda"] },
+            "FunctionName": { "type": "string" },
+            "InvokeArgs": { "type": "object" }
+          },
+          "required": ["type", "FunctionName", "InvokeArgs"]
+        },
+        {
+          "properties": {
+            "type": { "type": "string", "enum": ["http"] },
+            "method": { "type": "string", "enum": ["GET", "get", "Get", "POST", "post", "Post"]},
+            "request": { "type": "object" }
+          },
+          "required": ["type", "method", "request"]
+        }
+      ]
+    }
+  },
+  "required": ["name", "expression", "task"]
+}

tests/resources/tasks/invalid/invalid_task.yml

@@ -0,0 +1,8 @@
+name: 'test queue task'
+expression: '0 a * * *'
+task:
+  type: 'queue'
+  QueueName: 'test-queue'
+  MessageBody:
+    message_key_1: 'message_value_1'
+    message_key_2: ['message_value_2']

tests/resources/tasks/valid/http_get_task.yml

@@ -0,0 +1,10 @@
+name: 'test http get task'
+expression: '0 11 * * *'
+task:
+  type: 'http'
+  method: 'GET'
+  request:
+    url: 'http://lambda-cron.com/tests'
+    params:
+      param_1: 'param_value_1'
+      param_2: 2

tests/resources/tasks/valid/http_post_task.yml

@@ -0,0 +1,10 @@
+name: 'test http post task'
+expression: '0 11 * * *'
+task:
+  type: 'http'
+  method: 'POST'
+  request:
+    url: 'http://lambda-cron.com/tests'
+    data:
+      post_param_1: 'param_value_1'
+      post_param_2: 2

tests/resources/tasks/valid/lambda_task.yml

@@ -0,0 +1,8 @@
+name: 'test lambda task'
+expression: '0 11 * * *'
+task:
+  type: 'lambda'
+  FunctionName: 'testing-function'
+  InvokeArgs:
+    argument_1: 'value_argument_1'
+    argument_2: 2

tests/resources/tasks/valid/queue_task.yml

@@ -0,0 +1,8 @@
+name: 'test queue task'
+expression: '0 11 * * *'
+task:
+  type: 'queue'
+  QueueName: 'test-queue'
+  MessageBody:
+    message_key_1: 'message_value_1'
+    message_key_2: ['message_value_2']