Merge 6981164 into 4b58321

chroma_feedback/producer/__init__.py

@@ -10,6 +10,7 @@
 	'circle',
 	'codeship',
 	'custom',
+	'datadog',
 	'github',
 	'gitlab',
 	'jenkins',

chroma_feedback/producer/datadog/__init__.py

@@ -0,0 +1 @@
+from .core import init, run

chroma_feedback/producer/datadog/core.py

@@ -0,0 +1,49 @@
+from typing import List
+from argparse import ArgumentParser
+from chroma_feedback import helper, request
+from chroma_feedback.typing import Producer
+from .normalize import normalize_data
+
+ARGS = None
+
+
+def init(program : ArgumentParser) -> None:
+	global ARGS
+
+	if not ARGS:
+		program.add_argument('--datadog-host', default = 'https://api.datadoghq.com')
+		program.add_argument('--datadog-slug', action = 'append', required = True)
+		program.add_argument('--datadog-api-key', required = True)
+		program.add_argument('--datadog-application-key', required = True)
+	ARGS = helper.get_first(program.parse_known_args())
+
+
+def run() -> List[Producer]:
+	result = []
+
+	if ARGS.datadog_slug:
+		for slug in ARGS.datadog_slug:
+			result.extend(fetch(ARGS.azure_host, slug, ARGS.datadog_api_key, ARGS.datadog_application_key))
+	return result
+
+
+def fetch(host : str, slug : str, api_key : str, application_key : str) -> List[Producer]:
+	result = []
+	response = None
+
+	if host and slug and api_key and application_key:
+		response = request.get(host + '/api/v1/monitor/' + slug, headers =
+		{
+			'Accept': 'application/json',
+			'DD-API-KEY': api_key,
+			'DD-APPLICATION-KEY': application_key
+		})
+
+	# process response
+
+	if response and response.status_code == 200:
+		data = request.parse_json(response)
+
+		if 'name' in data and 'overall_state' in data:
+			result.append(normalize_data(data['name'], data['overall_state']))
+	return result

chroma_feedback/producer/datadog/normalize.py

@@ -0,0 +1,22 @@
+from chroma_feedback import helper
+from chroma_feedback.typing import Status, Producer
+
+
+def normalize_data(slug : str, status : str) -> Producer:
+	return\
+	{
+		'producer': 'datadog',
+		'slug': slug,
+		'active': True,
+		'status': normalize_status(status)
+	}
+
+
+def normalize_status(status : str) -> Status:
+	status = helper.to_lower_case(status)
+
+	if status in ['warn']:
+		return 'errored'
+	if status == 'alert':
+		return 'failed'
+	return 'passed'

tests/producer/datadog/test_core.py

@@ -0,0 +1,25 @@
+from typing import Any
+from unittest.mock import patch
+from chroma_feedback.producer.datadog.core import fetch
+
+
+@patch('requests.get')
+def test_fetch_slug(request_mock : Any) -> None:
+	request_mock.return_value.status_code = 200
+	request_mock.return_value.json.return_value =\
+	{
+		'name': 'chroma-feedback',
+		'overall_state': 'OK'
+	}
+	result = fetch('https://api.datadoghq.com', '1', 'api_key', 'application_key')
+
+	assert result[0]['producer'] == 'datadog'
+	assert result[0]['slug'] == 'chroma-feedback'
+	assert result[0]['active'] is True
+	assert result[0]['status']
+
+
+def test_fetch_invalid() -> None:
+	result = fetch(None, None, None, None)
+
+	assert result == []