1import json
2
3import httpretty
4
5from pypuppetdb.types import Node, Inventory, Fact
6
7
8class TestPqlAPI(object):
9
10    def test_pql_casting(self, api):
11        pql_query = """
12          nodes {
13            facts {
14              name = "operatingsystem" and
15              value = "Debian"
16            }
17          }
18        """
19        pql_body = [
20            {
21                "cached_catalog_status": "not_used",
22                "catalog_environment": "production",
23                "catalog_timestamp": "2016-08-15T11:06:26.275Z",
24                "certname": "greenserver.vm",
25                "deactivated": None,
26                "expired": None,
27                "facts_environment": "production",
28                "facts_timestamp": "2016-08-15T11:06:26.140Z",
29                "latest_report_hash": "4a956674b016d95a7b77c99513ba26e4a744f8d1",
30                "latest_report_noop": False,
31                "latest_report_noop_pending": None,
32                "latest_report_status": "changed",
33                "report_environment": "production",
34                "report_timestamp": "2016-08-15T11:06:18.393Z"
35            },
36            {
37                "cached_catalog_status": "not_used",
38                "catalog_environment": "production",
39                "catalog_timestamp": "2016-08-15T11:06:26.275Z",
40                "certname": "blueserver.vm",
41                "deactivated": None,
42                "expired": None,
43                "facts_environment": "production",
44                "facts_timestamp": "2016-08-15T11:06:26.140Z",
45                "latest_report_hash": "4a956674b016d95a7b77c99513ba26e4a744f8d1",
46                "latest_report_noop": False,
47                "latest_report_noop_pending": None,
48                "latest_report_status": "changed",
49                "report_environment": "production",
50                "report_timestamp": "2016-08-15T11:06:18.393Z"
51            }
52        ]
53        pql_url = 'http://localhost:8080/pdb/query/v4'
54
55        httpretty.enable()
56        httpretty.register_uri(httpretty.GET, pql_url,
57                               body=json.dumps(pql_body))
58
59        nodes = list(api.pql(pql_query))
60
61        assert type(nodes[0]) == Node
62        assert nodes[0].name == "greenserver.vm"
63
64        httpretty.disable()
65        httpretty.reset()
66
67    def test_pql_no_casting(self, api):
68        pql_query = """
69          nodes[certname] {
70            facts {
71              name = "operatingsystem" and
72              value = "Debian"
73            }
74          }
75        """
76        pql_body = [
77            {'certname': 'foo.example.com'},
78            {'certname': 'bar.example.com'},
79        ]
80        pql_url = 'http://localhost:8080/pdb/query/v4'
81
82        httpretty.enable()
83        httpretty.register_uri(httpretty.GET, pql_url,
84                               body=json.dumps(pql_body))
85
86        elements = list(api.pql(pql_query))
87
88        assert httpretty.last_request().path.startswith('/pdb/query/v4')
89        assert elements[0]["certname"] == 'foo.example.com'
90
91        httpretty.disable()
92        httpretty.reset()
93
94    def test_get_type_from_query_matching(self, api):
95        pql = """
96              nodes {
97                facts {
98                  name = "operatingsystem" and
99                  value = "Debian"
100                }
101              }
102            """
103        type = api._get_type_from_query(pql)
104
105        assert type == Node
106
107    def test_get_type_from_query_matching_empty_projection(self, api):
108        pql = """
109              inventory[] {}
110            """
111        type = api._get_type_from_query(pql)
112
113        assert type == Inventory
114
115    def test_get_type_from_query_matching_no_whitespace(self, api):
116        pql = """
117              facts{}
118            """
119        type = api._get_type_from_query(pql)
120
121        assert type == Fact
122
123    def test_get_type_from_query_matching_but_unsupported(self, api):
124        pql = """
125              producers {}
126            """
127        type = api._get_type_from_query(pql)
128
129        assert type is None
130
131    def test_get_type_from_query_not_matching_projection(self, api):
132        pql = """
133              nodes[certname, count()] {}
134            """
135        type = api._get_type_from_query(pql)
136
137        assert type is None
138