xp-dev.com Git - nip.io/blob - nipio_tests/backend_test.py

   1 # Copyright 2019 Exentrique Solutions Ltd
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import collections
  16 import os
  17 import sys
  18 import unittest
  19
  20 from assertpy import assert_that
  21 from mock.mock import patch, call
  22
  23 from nipio.backend import DynamicBackend
  24
  25
  26 class DynamicBackendTest(unittest.TestCase):
  27     def setUp(self):
  28         self.mock_sys_patcher = patch("nipio.backend.sys")
  29         self.mock_sys = self.mock_sys_patcher.start()
  30
  31         self.mock_sys.stderr.write = sys.stderr.write
  32
  33         import nipio
  34         nipio.backend._is_debug = lambda: True
  35
  36     def tearDown(self):
  37         sys.stderr.flush()
  38
  39         self.mock_sys_patcher.stop()
  40
  41     def test_backend_ends_response_to_ANY_request_if_ip_is_blacklisted(self):
  42         self._send_commands(["Q", "subdomain.127.0.0.2.lcl.io", "IN", "ANY", "1", "127.0.0.1"])
  43
  44         self._run_backend()
  45
  46         self._assert_expected_responses(["LOG", "Blacklisted: 127.0.0.2"])
  47
  48     def test_backend_ends_response_to_A_request_if_ip_is_blacklisted(self):
  49         self._send_commands(["Q", "subdomain.127.0.0.2.lcl.io", "IN", "A", "1", "127.0.0.1"])
  50
  51         self._run_backend()
  52
  53         self._assert_expected_responses(
  54             ["LOG", "Blacklisted: 127.0.0.2"]
  55         )
  56
  57     def test_backend_responds_to_ANY_request_with_valid_ip(self):
  58         self._send_commands(["Q", "subdomain.127.0.0.1.lcl.io", "IN", "ANY", "1", "127.0.0.1"])
  59
  60         self._run_backend()
  61
  62         self._assert_expected_responses(
  63             ["DATA", "subdomain.127.0.0.1.lcl.io", "IN", "A", "200", "22", "127.0.0.1"],
  64             ["DATA", "subdomain.127.0.0.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
  65             ["DATA", "subdomain.127.0.0.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
  66         )
  67
  68     def test_backend_responds_to_A_request_with_valid_ip(self):
  69         self._send_commands(["Q", "subdomain.127.0.0.1.lcl.io", "IN", "A", "1", "127.0.0.1"])
  70
  71         self._run_backend()
  72
  73         self._assert_expected_responses(
  74             ["DATA", "subdomain.127.0.0.1.lcl.io", "IN", "A", "200", "22", "127.0.0.1"],
  75             ["DATA", "subdomain.127.0.0.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
  76             ["DATA", "subdomain.127.0.0.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
  77         )
  78
  79     def test_backend_responds_to_invalid_ip_in_ANY_request_with_self_ip(self):
  80         self._send_commands(["Q", "subdomain.127.0.1.lcl.io", "IN", "ANY", "1", "127.0.0.1"])
  81
  82         self._run_backend()
  83
  84         self._assert_expected_responses(
  85             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
  86             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
  87             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
  88         )
  89
  90     def test_backend_responds_to_invalid_ip_in_A_request_with_self(self):
  91         self._send_commands(["Q", "subdomain.127.0.1.lcl.io", "IN", "A", "1", "127.0.0.1"])
  92
  93         self._run_backend()
  94
  95         self._assert_expected_responses(
  96             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
  97             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
  98             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
  99         )
 100
 101     def test_backend_responds_to_short_ip_in_ANY_request_with_self_ip(self):
 102         self._send_commands(["Q", "127.0.1.lcl.io", "IN", "ANY", "1", "127.0.0.1"])
 103
 104         self._run_backend()
 105
 106         self._assert_expected_responses(
 107             ["DATA", "127.0.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 108             ["DATA", "127.0.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 109             ["DATA", "127.0.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 110         )
 111
 112     def test_backend_responds_to_short_ip_in_A_request_with_self(self):
 113         self._send_commands(["Q", "127.0.1.lcl.io", "IN", "A", "1", "127.0.0.1"])
 114
 115         self._run_backend()
 116
 117         self._assert_expected_responses(
 118             ["DATA", "127.0.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 119             ["DATA", "127.0.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 120             ["DATA", "127.0.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 121         )
 122
 123     def test_backend_responds_to_large_ip_in_ANY_request_with_self(self):
 124         self._send_commands(["Q", "subdomain.127.0.300.1.lcl.io", "IN", "ANY", "1", "127.0.0.1"])
 125
 126         self._run_backend()
 127
 128         self._assert_expected_responses(
 129             ["DATA", "subdomain.127.0.300.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 130             ["DATA", "subdomain.127.0.300.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 131             ["DATA", "subdomain.127.0.300.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 132         )
 133
 134     def test_backend_responds_to_large_ip_in_A_request_with_self(self):
 135         self._send_commands(["Q", "subdomain.127.0.300.1.lcl.io", "IN", "A", "1", "127.0.0.1"])
 136
 137         self._run_backend()
 138
 139         self._assert_expected_responses(
 140             ["DATA", "subdomain.127.0.300.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 141             ["DATA", "subdomain.127.0.300.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 142             ["DATA", "subdomain.127.0.300.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 143         )
 144
 145     def test_backend_responds_to_string_in_ip_in_ANY_request_with_self(self):
 146         self._send_commands(["Q", "subdomain.127.0.STRING.1.lcl.io", "IN", "ANY", "1", "127.0.0.1"])
 147
 148         self._run_backend()
 149
 150         self._assert_expected_responses(
 151             ["DATA", "subdomain.127.0.string.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 152             ["DATA", "subdomain.127.0.string.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 153             ["DATA", "subdomain.127.0.string.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 154         )
 155
 156     def test_backend_responds_to_string_in_ip_in_A_request_with_self(self):
 157         self._send_commands(["Q", "subdomain.127.0.STRING.1.lcl.io", "IN", "A", "1", "127.0.0.1"])
 158
 159         self._run_backend()
 160
 161         self._assert_expected_responses(
 162             ["DATA", "subdomain.127.0.string.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 163             ["DATA", "subdomain.127.0.string.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 164             ["DATA", "subdomain.127.0.string.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 165         )
 166
 167     def test_backend_responds_to_no_ip_in_ANY_request_with_self(self):
 168         self._send_commands(["Q", "subdomain.127.0.1.lcl.io", "IN", "ANY", "1", "127.0.0.1"])
 169
 170         self._run_backend()
 171
 172         self._assert_expected_responses(
 173             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 174             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 175             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 176         )
 177
 178     def test_backend_responds_to_no_ip_in_A_request_with_self(self):
 179         self._send_commands(["Q", "subdomain.127.0.1.lcl.io", "IN", "A", "1", "127.0.0.1"])
 180
 181         self._run_backend()
 182
 183         self._assert_expected_responses(
 184             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 185             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 186             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 187         )
 188
 189     def test_backend_responds_to_self_domain_to_A_request(self):
 190         self._send_commands(["Q", "lcl.io", "IN", "A", "1", "127.0.0.1"])
 191
 192         self._run_backend()
 193
 194         self._assert_expected_responses(
 195             ["DATA", "lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 196             ["DATA", "lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 197             ["DATA", "lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 198         )
 199
 200     def test_backend_responds_to_self_domain_to_ANY_request(self):
 201         self._send_commands(["Q", "lcl.io", "IN", "ANY", "1", "127.0.0.1"])
 202
 203         self._run_backend()
 204
 205         self._assert_expected_responses(
 206             ["DATA", "lcl.io", "IN", "A", "200", "22", "127.0.0.33"],
 207             ["DATA", "lcl.io", "IN", "NS", "200", "22", "ns1.lcl.io"],
 208             ["DATA", "lcl.io", "IN", "NS", "200", "22", "ns2.lcl.io"],
 209         )
 210
 211     def test_backend_responds_to_name_servers_A_request_with_valid_ip(self):
 212         self._send_commands(["Q", "ns1.lcl.io", "IN", "A", "1", "127.0.0.1"])
 213
 214         self._run_backend()
 215
 216         self._assert_expected_responses(
 217             ["DATA", "ns1.lcl.io", "IN", "A", "200", "22", "127.0.0.34"],
 218         )
 219
 220     def test_backend_responds_to_name_servers_ANY_request_with_valid_ip(self):
 221         self._send_commands(["Q", "ns2.lcl.io", "IN", "ANY", "1", "127.0.0.1"])
 222
 223         self._run_backend()
 224
 225         self._assert_expected_responses(
 226             ["DATA", "ns2.lcl.io", "IN", "A", "200", "22", "127.0.0.35"],
 227         )
 228
 229     def test_backend_responds_to_SOA_request_for_self(self):
 230         self._send_commands(["Q", "lcl.io", "IN", "SOA", "1", "127.0.0.1"])
 231
 232         self._run_backend()
 233
 234         self._assert_expected_responses(
 235             ["DATA", "lcl.io", "IN", "SOA", "200", "22", "MY_SOA"]
 236         )
 237
 238     def test_backend_responds_to_SOA_request_for_valid_ip(self):
 239         self._send_commands(["Q", "subdomain.127.0.0.1.lcl.io", "IN", "SOA", "1", "127.0.0.1"])
 240
 241         self._run_backend()
 242
 243         self._assert_expected_responses(
 244             ["DATA", "subdomain.127.0.0.1.lcl.io", "IN", "SOA", "200", "22", "MY_SOA"]
 245         )
 246
 247     def test_backend_responds_to_SOA_request_for_invalid_ip(self):
 248         self._send_commands(["Q", "subdomain.127.0.1.lcl.io", "IN", "SOA", "1", "127.0.0.1"])
 249
 250         self._run_backend()
 251
 252         self._assert_expected_responses(
 253             ["DATA", "subdomain.127.0.1.lcl.io", "IN", "SOA", "200", "22", "MY_SOA"]
 254         )
 255
 256     def test_backend_responds_to_SOA_request_for_no_ip(self):
 257         self._send_commands(["Q", "subdomain.lcl.io", "IN", "SOA", "1", "127.0.0.1"])
 258
 259         self._run_backend()
 260
 261         self._assert_expected_responses(
 262             ["DATA", "subdomain.lcl.io", "IN", "SOA", "200", "22", "MY_SOA"]
 263         )
 264
 265     def test_backend_responds_to_SOA_request_for_nameserver(self):
 266         self._send_commands(["Q", "ns1.lcl.io", "IN", "SOA", "1", "127.0.0.1"])
 267
 268         self._run_backend()
 269
 270         self._assert_expected_responses(
 271             ["DATA", "ns1.lcl.io", "IN", "SOA", "200", "22", "MY_SOA"]
 272         )
 273
 274     def test_backend_responds_to_A_request_for_unknown_domain_with_invalid_response(self):
 275         self._send_commands(["Q", "unknown.domain", "IN", "A", "1", "127.0.0.1"])
 276
 277         self._run_backend()
 278
 279         self._assert_expected_responses(
 280             ["LOG", "Unknown type: A, domain: unknown.domain"]
 281         )
 282
 283     def test_backend_responds_to_invalid_request_with_invalid_response(self):
 284         self._send_commands(["Q", "lcl.io", "IN", "INVALID", "1", "127.0.0.1"])
 285
 286         self._run_backend()
 287
 288         self._assert_expected_responses(
 289             ["LOG", "Unknown type: INVALID, domain: lcl.io"]
 290         )
 291
 292     def test_backend_responds_to_invalid_command_with_fail(self):
 293         self._send_commands(["INVALID", "COMMAND"])
 294
 295         self._run_backend()
 296
 297         calls = [
 298             call("OK"),
 299             call("\t"),
 300             call("We are good"),
 301             call("\n"),
 302
 303             call("FAIL"),
 304             call("\n"),
 305         ]
 306
 307         self.mock_sys.stdout.write.assert_has_calls(calls)
 308         assert_that(self.mock_sys.stdout.write.call_count).is_equal_to(len(calls))
 309
 310         assert_that(self.mock_sys.stdout.flush.call_count).is_equal_to(2)
 311
 312     def test_configure_with_full_config(self):
 313         backend = self._configure_backend()
 314
 315         assert_that(backend.id).is_equal_to("55")
 316         assert_that(backend.ip_address).is_equal_to("127.0.0.40")
 317         assert_that(backend.domain).is_equal_to("lcl.io")
 318         assert_that(backend.ttl).is_equal_to("1000")
 319         assert_that(backend.name_servers).is_equal_to({"ns1.lcl.io": "127.0.0.41", "ns2.lcl.io" : "127.0.0.42"})
 320         assert_that(backend.blacklisted_ips).is_equal_to(["10.0.0.100"])
 321         assert_that(backend.soa).is_equal_to("ns1.lcl.io emailaddress@lcl.io 55")
 322
 323     def test_configure_with_config_missing_blacklists(self):
 324         backend = self._configure_backend(filename="backend_test_no_blacklist.conf")
 325
 326         assert_that(backend.blacklisted_ips).is_empty()
 327
 328     def _run_backend(self):
 329         backend = self._create_backend()
 330         backend.run()
 331
 332     def _send_commands(self, *commands):
 333         commands_to_send = ["HELO\t1\n"]
 334
 335         for command in commands:
 336             commands_to_send.append("\t".join(command) + "\n")
 337
 338         commands_to_send.append("END\n")
 339
 340         self.mock_sys.stdin.readline.side_effect = commands_to_send
 341
 342     def _assert_expected_responses(self, *responses):
 343         calls = [
 344             call("OK"),
 345             call("\t"),
 346             call("We are good"),
 347             call("\n"),
 348         ]
 349
 350         for response in responses:
 351             tab_separated = ["\t"] * (len(response) * 2 - 1)
 352             tab_separated[0::2] = response
 353             tab_separated.append("\n")
 354
 355             calls.extend([call(response_item) for response_item in tab_separated])
 356
 357         calls.extend([
 358             call("END"),
 359             call("\n"),
 360         ])
 361
 362         self.mock_sys.stdout.write.assert_has_calls(calls)
 363         assert_that(self.mock_sys.stdout.write.call_count).is_equal_to(len(calls))
 364
 365         assert_that(self.mock_sys.stdout.flush.call_count).is_equal_to(len(responses) + 2)
 366
 367     @staticmethod
 368     def _create_backend():
 369         backend = DynamicBackend()
 370         backend.id = "22"
 371         backend.soa = "MY_SOA"
 372         backend.ip_address = "127.0.0.33"
 373         backend.ttl = "200"
 374         backend.name_servers = collections.OrderedDict([
 375             ("ns1.lcl.io", "127.0.0.34"),
 376             ("ns2.lcl.io", "127.0.0.35"),
 377         ])
 378         backend.domain = "lcl.io"
 379         backend.blacklisted_ips = ["127.0.0.2"]
 380         return backend
 381
 382     def _configure_backend(self, filename="backend_test.conf"):
 383         backend = DynamicBackend()
 384         backend._get_config_filename = lambda: self._get_test_config_filename(filename)
 385         backend.configure()
 386         return backend
 387
 388     def _get_test_config_filename(self, filename):
 389         return os.path.join(os.path.dirname(os.path.realpath(__file__)), filename)
 390