Skip to content

Commit

Permalink
Split db tests (#9814)
Browse files Browse the repository at this point in the history 
* Split db tests
  • Loading branch information
@hithwen
hithwen authored Aug 2, 2021
1 parent 27886ea commit a2b6cd8
Show file tree
Hide file tree
Showing 7 changed files with 875 additions and 840 deletions.
Empty file added datadog_checks_base/tests/base/utils/db/__init__.py
View file
Empty file.
23 changes: 23 additions & 0 deletions datadog_checks_base/tests/base/utils/db/common.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
# (C) Datadog, Inc. 2021-present
# All rights reserved
# Licensed under a 3-clause BSD style license (see LICENSE)
from datadog_checks.base import AgentCheck
from datadog_checks.base.utils.db import QueryManager


def mock_executor(result=()):
def executor(_):
return result

return executor


def create_query_manager(*args, **kwargs):
executor = kwargs.pop('executor', None)
if executor is None:
executor = mock_executor()

check = kwargs.pop('check', None) or AgentCheck('test', {}, [{}])
check.check_id = 'test:instance'

return QueryManager(check, executor, args, **kwargs)
185 changes: 185 additions & 0 deletions datadog_checks_base/tests/base/utils/db/test_custom_queries.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,185 @@
# (C) Datadog, Inc. 2021-present
# All rights reserved
# Licensed under a 3-clause BSD style license (see LICENSE)
import pytest

from datadog_checks.base import AgentCheck

from .common import create_query_manager, mock_executor

pytestmark = pytest.mark.db


class TestCustomQueries:
def test_instance(self, aggregator):
query_manager = create_query_manager(
check=AgentCheck(
'test',
{},
[
{
'custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.foo', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
],
),
executor=mock_executor([[1]]),
tags=['test:foo'],
)
query_manager.compile_queries()
query_manager.execute()

aggregator.assert_metric('test.foo', 1, metric_type=aggregator.GAUGE, tags=['test:foo', 'test:bar'])
aggregator.assert_all_metrics_covered()

def test_init_config(self, aggregator):
query_manager = create_query_manager(
check=AgentCheck(
'test',
{
'global_custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.foo', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
[{}],
),
executor=mock_executor([[1]]),
tags=['test:foo'],
)
query_manager.compile_queries()
query_manager.execute()

aggregator.assert_metric('test.foo', 1, metric_type=aggregator.GAUGE, tags=['test:foo', 'test:bar'])
aggregator.assert_all_metrics_covered()

def test_instance_override(self, aggregator):
query_manager = create_query_manager(
check=AgentCheck(
'test',
{
'global_custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.foo', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
[
{
'custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.bar', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
],
),
executor=mock_executor([[1]]),
tags=['test:foo'],
)
query_manager.compile_queries()
query_manager.execute()

aggregator.assert_metric('test.bar', 1, metric_type=aggregator.GAUGE, tags=['test:foo', 'test:bar'])
aggregator.assert_all_metrics_covered()

def test_global_include(self, aggregator):
query_manager = create_query_manager(
check=AgentCheck(
'test',
{
'global_custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.foo', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
[
{
'use_global_custom_queries': 'extend',
'custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.bar', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
],
),
executor=mock_executor([[1]]),
tags=['test:foo'],
)
query_manager.compile_queries()
query_manager.execute()

aggregator.assert_metric('test.foo', 1, metric_type=aggregator.GAUGE, tags=['test:foo', 'test:bar'])
aggregator.assert_metric('test.bar', 1, metric_type=aggregator.GAUGE, tags=['test:foo', 'test:bar'])
aggregator.assert_all_metrics_covered()

def test_global_exclude(self, aggregator):
query_manager = create_query_manager(
check=AgentCheck(
'test',
{
'global_custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.foo', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
[
{
'use_global_custom_queries': False,
'custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.bar', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
],
),
executor=mock_executor([[1]]),
tags=['test:foo'],
)
query_manager.compile_queries()
query_manager.execute()

aggregator.assert_metric('test.bar', 1, metric_type=aggregator.GAUGE, tags=['test:foo', 'test:bar'])
aggregator.assert_all_metrics_covered()

def test_deduplication(self, aggregator):
query_manager = create_query_manager(
check=AgentCheck(
'test',
{
'global_custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.foo', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
[
{
'use_global_custom_queries': 'extend',
'custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.foo', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
],
),
executor=mock_executor([[1]]),
tags=['test:foo'],
)
query_manager.compile_queries()
query_manager.execute()

aggregator.assert_metric('test.foo', 1, count=1, metric_type=aggregator.GAUGE, tags=['test:foo', 'test:bar'])
aggregator.assert_all_metrics_covered()

def test_default_name(self):
query_manager = create_query_manager(
check=AgentCheck(
'test',
{
'global_custom_queries': [
{'query': 'foo', 'columns': [{'name': 'test.foo', 'type': 'gauge'}], 'tags': ['test:bar']},
],
},
[
{
'use_global_custom_queries': 'extend',
'custom_queries': [{'columns': [{'name': 'test.bar', 'type': 'gauge'}], 'tags': ['test:bar']}],
},
],
),
executor=mock_executor([[1]]),
tags=['test:foo'],
)

with pytest.raises(ValueError, match='^field `query` for custom query #1 is required$'):
query_manager.compile_queries()
Loading

0 comments on commit a2b6cd8

Please sign in to comment.