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