test_rrsets_bulk.py 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739
  1. import copy
  2. from django.conf import settings
  3. from rest_framework import status
  4. from desecapi.tests.base import AuthenticatedRRSetBaseTestCase
  5. class AuthenticatedRRSetBulkTestCase(AuthenticatedRRSetBaseTestCase):
  6. @classmethod
  7. def setUpTestDataWithPdns(cls):
  8. super().setUpTestDataWithPdns()
  9. cls.data = [
  10. {
  11. "subname": "my-cname",
  12. "records": ["example.com."],
  13. "ttl": 3600,
  14. "type": "CNAME",
  15. },
  16. {
  17. "subname": "my-bulk",
  18. "records": ["desec.io.", "foobar.example."],
  19. "ttl": 3600,
  20. "type": "PTR",
  21. },
  22. ]
  23. cls.data_no_records = copy.deepcopy(cls.data)
  24. cls.data_no_records[1].pop("records")
  25. cls.data_empty_records = copy.deepcopy(cls.data)
  26. cls.data_empty_records[1]["records"] = []
  27. cls.data_no_subname = copy.deepcopy(cls.data)
  28. cls.data_no_subname[0].pop("subname")
  29. cls.data_no_ttl = copy.deepcopy(cls.data)
  30. cls.data_no_ttl[0].pop("ttl")
  31. cls.data_no_type = copy.deepcopy(cls.data)
  32. cls.data_no_type[1].pop("type")
  33. cls.data_no_records_no_ttl = copy.deepcopy(cls.data_no_records)
  34. cls.data_no_records_no_ttl[1].pop("ttl")
  35. cls.data_no_subname_empty_records = copy.deepcopy(cls.data_no_subname)
  36. cls.data_no_subname_empty_records[0]["records"] = []
  37. cls.bulk_domain = cls.create_domain(owner=cls.owner)
  38. for data in cls.data:
  39. cls.create_rr_set(cls.bulk_domain, **data)
  40. def test_bulk_post_my_rr_sets(self):
  41. with self.assertPdnsRequests(
  42. self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
  43. ):
  44. response = self.client.bulk_post_rr_sets(
  45. domain_name=self.my_empty_domain.name, payload=self.data
  46. )
  47. self.assertStatus(response, status.HTTP_201_CREATED)
  48. response = self.client.get_rr_sets(self.my_empty_domain.name)
  49. self.assertStatus(response, status.HTTP_200_OK)
  50. self.assertRRSetsCount(response.data, self.data)
  51. # Check subname requirement on bulk endpoint (and uniqueness at the same time)
  52. self.assertResponse(
  53. self.client.bulk_post_rr_sets(
  54. domain_name=self.my_empty_domain.name, payload=self.data_no_subname
  55. ),
  56. status.HTTP_400_BAD_REQUEST,
  57. [
  58. {"subname": ["This field is required."]},
  59. {
  60. "non_field_errors": [
  61. "Another RRset with the same subdomain and type exists for this domain."
  62. ]
  63. },
  64. ],
  65. )
  66. def test_bulk_post_rr_sets_empty_records(self):
  67. expected_response_data = [copy.deepcopy(self.data_empty_records[0]), None]
  68. expected_response_data[0]["domain"] = self.my_empty_domain.name
  69. expected_response_data[0]["name"] = "%s.%s." % (
  70. self.data_empty_records[0]["subname"],
  71. self.my_empty_domain.name,
  72. )
  73. self.assertResponse(
  74. self.client.bulk_post_rr_sets(
  75. domain_name=self.my_empty_domain.name, payload=self.data_empty_records
  76. ),
  77. status.HTTP_400_BAD_REQUEST,
  78. [{}, {"records": ["This field must not be empty when using POST."]}],
  79. )
  80. def test_bulk_post_existing_rrsets(self):
  81. self.assertResponse(
  82. self.client.bulk_post_rr_sets(
  83. domain_name=self.bulk_domain,
  84. payload=self.data,
  85. ),
  86. status.HTTP_400_BAD_REQUEST,
  87. 2
  88. * [
  89. {
  90. "non_field_errors": [
  91. "Another RRset with the same subdomain and type exists for this domain."
  92. ]
  93. }
  94. ],
  95. )
  96. def test_bulk_post_duplicates(self):
  97. data = 2 * [self.data[0]] + [self.data[1]]
  98. self.assertResponse(
  99. self.client.bulk_post_rr_sets(
  100. domain_name=self.my_empty_domain.name, payload=data
  101. ),
  102. status.HTTP_400_BAD_REQUEST,
  103. [
  104. {
  105. "non_field_errors": [
  106. "Same subname and type as in position(s) 1, but must be unique."
  107. ]
  108. },
  109. {
  110. "non_field_errors": [
  111. "Same subname and type as in position(s) 0, but must be unique."
  112. ]
  113. },
  114. {},
  115. ],
  116. )
  117. data = 2 * [self.data[0]] + [self.data[1]] + [self.data[0]]
  118. self.assertResponse(
  119. self.client.bulk_post_rr_sets(
  120. domain_name=self.my_empty_domain.name, payload=data
  121. ),
  122. status.HTTP_400_BAD_REQUEST,
  123. [
  124. {
  125. "non_field_errors": [
  126. "Same subname and type as in position(s) 1, 3, but must be unique."
  127. ]
  128. },
  129. {
  130. "non_field_errors": [
  131. "Same subname and type as in position(s) 0, 3, but must be unique."
  132. ]
  133. },
  134. {},
  135. {
  136. "non_field_errors": [
  137. "Same subname and type as in position(s) 0, 1, but must be unique."
  138. ]
  139. },
  140. ],
  141. )
  142. def test_bulk_post_missing_fields(self):
  143. self.assertResponse(
  144. self.client.bulk_post_rr_sets(
  145. domain_name=self.my_empty_domain.name,
  146. payload=[
  147. {"subname": "a.1", "records": ["dead::beef"], "ttl": 3622},
  148. {
  149. "subname": "b.1",
  150. "ttl": -50,
  151. "type": "AAAA",
  152. "records": ["dead::beef"],
  153. },
  154. {"ttl": 3640, "type": "TXT", "records": ['"bar"']},
  155. {"subname": "", "ttl": 3640, "type": "TXT", "records": ['"bar"']},
  156. {"subname": "c.1", "records": ["dead::beef"], "type": "AAAA"},
  157. {"subname": "d.1", "ttl": 3650, "type": "AAAA"},
  158. {
  159. "subname": "d.1",
  160. "ttl": 3650,
  161. "type": "SOA",
  162. "records": [
  163. "get.desec.io. get.desec.io. 2018034419 10800 3600 604800 60"
  164. ],
  165. },
  166. {"subname": "d.1", "ttl": 3650, "type": "OPT", "records": ["9999"]},
  167. {
  168. "subname": "d.1",
  169. "ttl": 3650,
  170. "type": "TYPE099",
  171. "records": ["v=spf1 mx -all"],
  172. },
  173. ],
  174. ),
  175. status.HTTP_400_BAD_REQUEST,
  176. [
  177. {"type": ["This field is required."]},
  178. {
  179. "ttl": [
  180. f"Ensure this value is greater than or equal to {self.my_empty_domain.minimum_ttl}."
  181. ]
  182. },
  183. {"subname": ["This field is required."]},
  184. {},
  185. {"ttl": ["This field is required."]},
  186. {"records": ["This field is required."]},
  187. {
  188. "type": [
  189. "You cannot tinker with the SOA RR set. It is managed automatically."
  190. ]
  191. },
  192. {
  193. "type": [
  194. "You cannot tinker with the OPT RR set. It is managed automatically."
  195. ]
  196. },
  197. {"type": ["Generic type format is not supported."]},
  198. ],
  199. )
  200. def test_bulk_patch_cname_exclusivity(self):
  201. response = self.client.bulk_patch_rr_sets(
  202. domain_name=self.my_rr_set_domain.name,
  203. payload=[
  204. {"subname": "test", "type": "A", "ttl": 3600, "records": ["1.2.3.4"]},
  205. {
  206. "subname": "test",
  207. "type": "CNAME",
  208. "ttl": 3600,
  209. "records": ["example.com."],
  210. },
  211. ],
  212. )
  213. self.assertResponse(response, status.HTTP_400_BAD_REQUEST)
  214. self.assertEqual(
  215. response.json(),
  216. [
  217. {
  218. "non_field_errors": [
  219. "RRset with conflicting type present: 1 (CNAME). (No other RRsets are allowed alongside CNAME.)"
  220. ]
  221. },
  222. {
  223. "non_field_errors": [
  224. "RRset with conflicting type present: 0 (A), database (A, TXT). (No other RRsets are allowed alongside CNAME.)"
  225. ]
  226. },
  227. ],
  228. )
  229. def test_bulk_post_accepts_empty_list(self):
  230. self.assertResponse(
  231. self.client.bulk_post_rr_sets(
  232. domain_name=self.my_empty_domain.name, payload=[]
  233. ),
  234. status.HTTP_201_CREATED,
  235. )
  236. def test_bulk_patch_fresh_rrsets_need_records(self):
  237. response = self.client.bulk_patch_rr_sets(
  238. self.my_empty_domain.name, payload=self.data_no_records
  239. )
  240. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  241. self.assertEqual(response.data, [{}, {"records": ["This field is required."]}])
  242. with self.assertPdnsRequests(
  243. self.requests_desec_rr_sets_update(self.my_empty_domain.name)
  244. ):
  245. response = self.client.bulk_patch_rr_sets(
  246. self.my_empty_domain.name, payload=self.data_empty_records
  247. )
  248. self.assertStatus(response, status.HTTP_200_OK)
  249. def test_bulk_patch_fresh_rrsets_need_subname(self):
  250. response = self.client.bulk_patch_rr_sets(
  251. domain_name=self.my_empty_domain.name, payload=self.data_no_subname
  252. )
  253. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  254. def test_bulk_patch_fresh_rrsets_need_ttl(self):
  255. response = self.client.bulk_patch_rr_sets(
  256. domain_name=self.my_empty_domain.name, payload=self.data_no_ttl
  257. )
  258. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  259. self.assertEqual(response.data, [{"ttl": ["This field is required."]}, {}])
  260. def test_bulk_patch_fresh_rrsets_need_type(self):
  261. response = self.client.bulk_patch_rr_sets(
  262. domain_name=self.my_empty_domain.name, payload=self.data_no_type
  263. )
  264. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  265. self.assertEqual(response.data, [{}, {"type": ["This field is required."]}])
  266. def test_bulk_patch_does_not_accept_single_objects(self):
  267. response = self.client.bulk_patch_rr_sets(
  268. domain_name=self.my_empty_domain.name, payload=self.data[0]
  269. )
  270. self.assertContains(
  271. response,
  272. "Expected a list of items but got dict.",
  273. status_code=status.HTTP_400_BAD_REQUEST,
  274. )
  275. def test_bulk_patch_does_accept_empty_list(self):
  276. response = self.client.bulk_patch_rr_sets(
  277. domain_name=self.my_empty_domain.name, payload=[]
  278. )
  279. self.assertStatus(response, status.HTTP_200_OK)
  280. response = self.client.bulk_patch_rr_sets(
  281. domain_name=self.my_rr_set_domain.name, payload=[]
  282. )
  283. self.assertStatus(response, status.HTTP_200_OK)
  284. def test_bulk_patch_does_not_accept_empty_payload(self):
  285. response = self.client.bulk_patch_rr_sets(
  286. domain_name=self.my_empty_domain.name, payload=None
  287. )
  288. self.assertContains(
  289. response, "No data provided", status_code=status.HTTP_400_BAD_REQUEST
  290. )
  291. def test_bulk_patch_cname_exclusivity_atomic_rrset_replacement(self):
  292. self.create_rr_set(
  293. self.my_empty_domain,
  294. subname="test",
  295. type="A",
  296. records=["1.2.3.4"],
  297. ttl=3600,
  298. )
  299. with self.assertPdnsRequests(
  300. self.requests_desec_rr_sets_update(self.my_empty_domain.name)
  301. ):
  302. response = self.client.bulk_patch_rr_sets(
  303. domain_name=self.my_empty_domain.name,
  304. payload=[
  305. {
  306. "subname": "test",
  307. "type": "CNAME",
  308. "ttl": 3605,
  309. "records": ["example.com."],
  310. },
  311. {"subname": "test", "type": "A", "records": []},
  312. ],
  313. )
  314. self.assertResponse(response, status.HTTP_200_OK)
  315. self.assertEqual(len(response.data), 1)
  316. self.assertEqual(response.data[0]["type"], "CNAME")
  317. self.assertEqual(response.data[0]["records"], ["example.com."])
  318. self.assertEqual(response.data[0]["ttl"], 3605)
  319. with self.assertPdnsRequests(
  320. self.requests_desec_rr_sets_update(self.my_empty_domain.name)
  321. ):
  322. response = self.client.bulk_patch_rr_sets(
  323. domain_name=self.my_empty_domain.name,
  324. payload=[
  325. {"subname": "test", "type": "CNAME", "records": []},
  326. {
  327. "subname": "test",
  328. "type": "A",
  329. "ttl": 3600,
  330. "records": ["5.4.2.1"],
  331. },
  332. ],
  333. )
  334. self.assertResponse(response, status.HTTP_200_OK)
  335. self.assertEqual(len(response.data), 1)
  336. self.assertEqual(response.data[0]["type"], "A")
  337. self.assertEqual(response.data[0]["records"], ["5.4.2.1"])
  338. self.assertEqual(response.data[0]["ttl"], 3600)
  339. def test_bulk_patch_full_on_empty_domain(self):
  340. # Full patch always works
  341. with self.assertPdnsRequests(
  342. self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
  343. ):
  344. response = self.client.bulk_patch_rr_sets(
  345. domain_name=self.my_empty_domain.name, payload=self.data
  346. )
  347. self.assertStatus(response, status.HTTP_200_OK)
  348. # Check that RRsets have been created
  349. response = self.client.get_rr_sets(self.my_empty_domain.name)
  350. self.assertStatus(response, status.HTTP_200_OK)
  351. self.assertRRSetsCount(response.data, self.data)
  352. def test_bulk_patch_change_records(self):
  353. data_no_ttl = copy.deepcopy(self.data_no_ttl)
  354. data_no_ttl[0]["records"] = ["example.org."]
  355. with self.assertPdnsRequests(
  356. self.requests_desec_rr_sets_update(name=self.bulk_domain.name)
  357. ):
  358. response = self.client.bulk_patch_rr_sets(
  359. domain_name=self.bulk_domain.name, payload=data_no_ttl
  360. )
  361. self.assertStatus(response, status.HTTP_200_OK)
  362. response = self.client.get_rr_sets(self.bulk_domain.name)
  363. self.assertStatus(response, status.HTTP_200_OK)
  364. self.assertRRSetsCount(response.data, data_no_ttl)
  365. def test_bulk_patch_change_ttl(self):
  366. data_no_records = copy.deepcopy(self.data_no_records)
  367. data_no_records[1]["ttl"] = 3911
  368. with self.assertPdnsRequests(
  369. self.requests_desec_rr_sets_update(name=self.bulk_domain.name)
  370. ):
  371. response = self.client.bulk_patch_rr_sets(
  372. domain_name=self.bulk_domain.name, payload=data_no_records
  373. )
  374. self.assertStatus(response, status.HTTP_200_OK)
  375. response = self.client.get_rr_sets(self.bulk_domain.name)
  376. self.assertStatus(response, status.HTTP_200_OK)
  377. self.assertRRSetsCount(response.data, data_no_records)
  378. def test_bulk_patch_does_not_need_ttl(self):
  379. self.assertResponse(
  380. self.client.bulk_patch_rr_sets(
  381. domain_name=self.bulk_domain.name, payload=self.data_no_ttl
  382. ),
  383. status.HTTP_200_OK,
  384. )
  385. def test_bulk_patch_delete_non_existing_rr_sets(self):
  386. self.assertResponse(
  387. self.client.bulk_patch_rr_sets(
  388. domain_name=self.my_empty_domain.name,
  389. payload=[
  390. {"subname": "a", "type": "A", "records": [], "ttl": 3622},
  391. {"subname": "b", "type": "AAAA", "records": []},
  392. ],
  393. ),
  394. status.HTTP_200_OK,
  395. [],
  396. )
  397. def test_bulk_patch_missing_invalid_fields_1(self):
  398. with self.assertPdnsRequests(
  399. self.requests_desec_rr_sets_update(self.my_empty_domain.name)
  400. ):
  401. self.client.bulk_post_rr_sets(
  402. domain_name=self.my_empty_domain.name,
  403. payload=[
  404. {"subname": "", "ttl": 3650, "type": "TXT", "records": ['"foo"']},
  405. {
  406. "subname": "c.1",
  407. "records": ["dead::beef"],
  408. "type": "AAAA",
  409. "ttl": 3603,
  410. },
  411. {
  412. "subname": "d.1",
  413. "ttl": 3650,
  414. "type": "AAAA",
  415. "records": ["::1", "::2"],
  416. },
  417. ],
  418. )
  419. self.assertResponse(
  420. self.client.bulk_patch_rr_sets(
  421. domain_name=self.my_empty_domain.name,
  422. payload=[
  423. {"subname": "a.1", "records": ["dead::beef"], "ttl": 3622},
  424. {
  425. "subname": "b.1",
  426. "ttl": -50,
  427. "type": "AAAA",
  428. "records": ["dead::beef"],
  429. },
  430. {"ttl": 3640, "type": "TXT", "records": ['"bar"']},
  431. {"subname": "c.1", "records": ["dead::beef"], "type": "AAAA"},
  432. {"subname": "d.1", "ttl": 3650, "type": "AAAA"},
  433. ],
  434. ),
  435. status.HTTP_400_BAD_REQUEST,
  436. [
  437. {"type": ["This field is required."]},
  438. {
  439. "ttl": [
  440. f"Ensure this value is greater than or equal to {settings.MINIMUM_TTL_DEFAULT}."
  441. ]
  442. },
  443. {"subname": ["This field is required."]},
  444. {},
  445. {},
  446. ],
  447. )
  448. def test_bulk_patch_missing_invalid_fields_2(self):
  449. with self.assertPdnsRequests(
  450. self.requests_desec_rr_sets_update(self.my_empty_domain.name)
  451. ):
  452. self.client.bulk_post_rr_sets(
  453. domain_name=self.my_empty_domain.name,
  454. payload=[
  455. {"subname": "", "ttl": 3650, "type": "TXT", "records": ['"foo"']}
  456. ],
  457. )
  458. self.assertResponse(
  459. self.client.bulk_patch_rr_sets(
  460. domain_name=self.my_empty_domain.name,
  461. payload=[
  462. {"subname": "a.1", "records": ["dead::beef"], "ttl": 3622},
  463. {
  464. "subname": "b.1",
  465. "ttl": -50,
  466. "type": "AAAA",
  467. "records": ["dead::beef"],
  468. },
  469. {"ttl": 3640, "type": "TXT", "records": ['"bar"']},
  470. {"subname": "c.1", "records": ["dead::beef"], "type": "AAAA"},
  471. {"subname": "d.1", "ttl": 3650, "type": "AAAA"},
  472. ],
  473. ),
  474. status.HTTP_400_BAD_REQUEST,
  475. [
  476. {"type": ["This field is required."]},
  477. {
  478. "ttl": [
  479. f"Ensure this value is greater than or equal to {settings.MINIMUM_TTL_DEFAULT}."
  480. ]
  481. },
  482. {"subname": ["This field is required."]},
  483. {"ttl": ["This field is required."]},
  484. {"records": ["This field is required."]},
  485. ],
  486. )
  487. def test_bulk_put_partial(self):
  488. # Need all fields
  489. for domain in [self.my_empty_domain, self.bulk_domain]:
  490. response = self.client.bulk_put_rr_sets(
  491. domain_name=domain.name, payload=self.data_no_records
  492. )
  493. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  494. self.assertEqual(
  495. response.data, [{}, {"records": ["This field is required."]}]
  496. )
  497. response = self.client.bulk_put_rr_sets(
  498. domain_name=domain.name, payload=self.data_no_ttl
  499. )
  500. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  501. self.assertEqual(response.data, [{"ttl": ["This field is required."]}, {}])
  502. response = self.client.bulk_put_rr_sets(
  503. domain_name=domain.name, payload=self.data_no_records_no_ttl
  504. )
  505. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  506. self.assertEqual(
  507. response.data,
  508. [
  509. {},
  510. {
  511. "ttl": ["This field is required."],
  512. "records": ["This field is required."],
  513. },
  514. ],
  515. )
  516. response = self.client.bulk_put_rr_sets(
  517. domain_name=domain.name, payload=self.data_no_subname
  518. )
  519. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  520. self.assertEqual(
  521. response.data, [{"subname": ["This field is required."]}, {}]
  522. )
  523. response = self.client.bulk_put_rr_sets(
  524. domain_name=domain.name, payload=self.data_no_subname_empty_records
  525. )
  526. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  527. self.assertEqual(
  528. response.data, [{"subname": ["This field is required."]}, {}]
  529. )
  530. response = self.client.bulk_put_rr_sets(
  531. domain_name=domain.name, payload=self.data_no_type
  532. )
  533. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  534. self.assertEqual(response.data, [{}, {"type": ["This field is required."]}])
  535. def test_bulk_put_does_not_accept_single_objects(self):
  536. response = self.client.bulk_put_rr_sets(
  537. domain_name=self.my_empty_domain.name, payload=self.data[0]
  538. )
  539. self.assertContains(
  540. response,
  541. "Expected a list of items but got dict.",
  542. status_code=status.HTTP_400_BAD_REQUEST,
  543. )
  544. def test_bulk_put_does_accept_empty_list(self):
  545. response = self.client.bulk_put_rr_sets(
  546. domain_name=self.my_empty_domain.name, payload=[]
  547. )
  548. self.assertStatus(response, status.HTTP_200_OK)
  549. response = self.client.bulk_put_rr_sets(
  550. domain_name=self.my_rr_set_domain.name, payload=[]
  551. )
  552. self.assertStatus(response, status.HTTP_200_OK)
  553. def test_bulk_put_does_not_accept_empty_payload(self):
  554. response = self.client.bulk_put_rr_sets(
  555. domain_name=self.my_empty_domain.name, payload=None
  556. )
  557. self.assertContains(
  558. response, "No data provided", status_code=status.HTTP_400_BAD_REQUEST
  559. )
  560. def test_bulk_put_does_not_accept_list_of_crap(self):
  561. response = self.client.bulk_put_rr_sets(
  562. domain_name=self.my_empty_domain.name, payload=["bla"]
  563. )
  564. self.assertContains(
  565. response,
  566. "Expected a dictionary, but got str.",
  567. status_code=status.HTTP_400_BAD_REQUEST,
  568. )
  569. response = self.client.bulk_put_rr_sets(
  570. domain_name=self.my_empty_domain.name, payload=[42]
  571. )
  572. self.assertContains(
  573. response,
  574. "Expected a dictionary, but got int.",
  575. status_code=status.HTTP_400_BAD_REQUEST,
  576. )
  577. def test_bulk_put_does_not_accept_rrsets_with_nonstr_subname(self):
  578. payload = [
  579. {"subname": ["foobar"], "type": "A", "ttl": 3600, "records": ["1.2.3.4"]}
  580. ]
  581. response = self.client.bulk_put_rr_sets(
  582. domain_name=self.my_empty_domain.name, payload=payload
  583. )
  584. self.assertContains(
  585. response,
  586. "Expected a string, but got list.",
  587. status_code=status.HTTP_400_BAD_REQUEST,
  588. )
  589. def test_bulk_put_does_not_accept_rrsets_with_nonstr_type(self):
  590. payload = [
  591. {"subname": "foobar", "type": ["A"], "ttl": 3600, "records": ["1.2.3.4"]}
  592. ]
  593. response = self.client.bulk_put_rr_sets(
  594. domain_name=self.my_empty_domain.name, payload=payload
  595. )
  596. self.assertContains(
  597. response,
  598. "Expected a string, but got list.",
  599. status_code=status.HTTP_400_BAD_REQUEST,
  600. )
  601. def test_bulk_put_full(self):
  602. # Full PUT always works
  603. with self.assertPdnsRequests(
  604. self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
  605. ):
  606. response = self.client.bulk_put_rr_sets(
  607. domain_name=self.my_empty_domain.name, payload=self.data
  608. )
  609. self.assertStatus(response, status.HTTP_200_OK)
  610. # Check that RRsets have been created
  611. response = self.client.get_rr_sets(self.my_empty_domain.name)
  612. self.assertStatus(response, status.HTTP_200_OK)
  613. self.assertRRSetsCount(response.data, self.data)
  614. # Do not expect any updates, but successful code when PUT'ing only existing RRsets
  615. response = self.client.bulk_put_rr_sets(
  616. domain_name=self.bulk_domain.name, payload=self.data
  617. )
  618. self.assertStatus(response, status.HTTP_200_OK)
  619. def test_bulk_put_invalid_records(self):
  620. for records in [
  621. "asfd",
  622. ["1.1.1.1", "2.2.2.2", 123],
  623. ["1.2.3.4", None],
  624. [True, "1.1.1.1"],
  625. dict(foobar="foobar", asdf="asdf"),
  626. ]:
  627. payload = [
  628. {"subname": "a.2", "ttl": 3600, "type": "MX", "records": records}
  629. ]
  630. response = self.client.bulk_put_rr_sets(
  631. domain_name=self.my_empty_domain.name, payload=payload
  632. )
  633. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  634. self.assertTrue("records" in response.data[0])
  635. def test_bulk_put_empty_records(self):
  636. with self.assertPdnsRequests(
  637. self.requests_desec_rr_sets_update(name=self.bulk_domain.name)
  638. ):
  639. self.assertStatus(
  640. self.client.bulk_put_rr_sets(
  641. domain_name=self.bulk_domain.name, payload=self.data_empty_records
  642. ),
  643. status.HTTP_200_OK,
  644. )
  645. def test_bulk_duplicate_rrset(self):
  646. data = self.data + self.data
  647. for bulk_request_rr_sets in [
  648. self.client.bulk_patch_rr_sets,
  649. self.client.bulk_put_rr_sets,
  650. self.client.bulk_post_rr_sets,
  651. ]:
  652. response = bulk_request_rr_sets(
  653. domain_name=self.my_empty_domain.name, payload=data
  654. )
  655. self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
  656. def test_bulk_patch_or_post_failure_with_single_rrset(self):
  657. for method in [self.client.bulk_patch_rr_sets, self.client.bulk_put_rr_sets]:
  658. response = method(
  659. domain_name=self.my_empty_domain.name, payload=self.data[0]
  660. )
  661. self.assertContains(
  662. response,
  663. "Expected a list of items but got dict.",
  664. status_code=status.HTTP_400_BAD_REQUEST,
  665. )
  666. def test_bulk_delete_rrsets(self):
  667. self.assertStatus(
  668. self.client.delete(
  669. self.reverse("v1:rrsets", name=self.my_empty_domain.name),
  670. data=None,
  671. ),
  672. status.HTTP_405_METHOD_NOT_ALLOWED,
  673. )