22
33from typing import TYPE_CHECKING , Any
44
5+ import pytest
6+ from infrahub_sdk .exceptions import GraphQLError
7+
58from tests .helpers .test_app import TestInfrahubApp
69
710if TYPE_CHECKING :
811 from infrahub_sdk import InfrahubClient
12+ from infrahub_sdk .node import InfrahubNode
913
1014
1115class TestDeleteAgnosticRel (TestInfrahubApp ):
16+ @pytest .fixture (scope = "class" )
17+ async def load_schema (self , client : InfrahubClient , car_person_branch_agnostic_schema : dict [str , Any ]) -> None :
18+ await client .schema .load ([car_person_branch_agnostic_schema ])
19+
20+ @pytest .fixture (scope = "class" )
21+ async def owner_1 (self , client : InfrahubClient , load_schema ) -> InfrahubNode :
22+ owner_1 = await client .create (kind = "TestPerson" , name = "owner_1" )
23+ await owner_1 .save ()
24+ return owner_1
25+
26+ @pytest .fixture (scope = "class" )
27+ async def owner_2 (self , client : InfrahubClient , load_schema ) -> InfrahubNode :
28+ owner_2 = await client .create (kind = "TestPerson" , name = "owner_2" )
29+ await owner_2 .save ()
30+ return owner_2
31+
32+ @pytest .fixture (scope = "class" )
33+ async def car (self , client : InfrahubClient , load_schema , owner_1 : InfrahubNode ) -> InfrahubNode :
34+ car = await client .create (kind = "TestCar" , name = "car_name" , owner = owner_1 )
35+ await car .save ()
36+ return car
37+
38+ @pytest .fixture (scope = "class" )
39+ async def car_2 (self , client : InfrahubClient , load_schema , owner_2 : InfrahubNode ) -> InfrahubNode :
40+ car = await client .create (kind = "TestCar" , name = "car_name_2" , owner = owner_2 )
41+ await car .save ()
42+ return car
43+
44+ @pytest .fixture (scope = "class" )
45+ async def roofrack_2 (self , client : InfrahubClient , load_schema , car_2 : InfrahubNode ) -> InfrahubNode :
46+ roofrack = await client .create (kind = "TestRoofrack" , size = "big" , car = car_2 )
47+ await roofrack .save ()
48+ return roofrack
49+
1250 async def test_delete_agnostic_rel (
13- self , client : InfrahubClient , car_person_branch_agnostic_schema : dict [str , Any ]
51+ self ,
52+ client : InfrahubClient ,
53+ load_schema ,
54+ owner_1 : InfrahubNode ,
55+ owner_2 : InfrahubNode ,
56+ car : InfrahubNode ,
1457 ) -> None :
1558 """
1659 Loads a car-person agnostic schema, then :
@@ -20,22 +63,42 @@ async def test_delete_agnostic_rel(
2063 This test makes sure changing owner, involving deleting relationship with first owner, works correctly.
2164 See https://github.com/opsmill/infrahub/issues/5559.
2265 """
66+ car = await client .get (kind = "TestCar" , name__value = "car_name" , prefetch_relationships = True )
67+ car .owner = owner_2
68+ await car .save ()
2369
24- await client .schema .load ([car_person_branch_agnostic_schema ])
70+ car = await client .get (kind = "TestCar" , name__value = "car_name" , prefetch_relationships = True )
71+ assert car .owner .peer .name .value == "owner_2"
2572
26- owner_1 = await client .create (kind = "TestPerson" , name = "owner_1" )
27- await owner_1 .save ()
73+ async def test_delete_aware_mandatory_node_blocked (
74+ self , client : InfrahubClient , owner_2 : InfrahubNode , car : InfrahubNode
75+ ) -> None :
76+ owner_2 = await client .get (kind = "TestPerson" , name__value = "owner_2" )
2877
29- car = await client .create (kind = "TestCar" , name = "car_name" , owner = owner_1 )
30- await car .save ()
78+ with pytest .raises (GraphQLError ) as exc :
79+ await owner_2 .delete ()
80+
81+ assert (
82+ f"Cannot delete TestPerson '{ owner_2 .id } '. It is linked to mandatory relationship owner on node TestCar '{ car .id } '"
83+ in exc .value .message
84+ )
3185
86+ async def test_delete_agnostic_node (self , client : InfrahubClient , owner_2 : InfrahubNode , car : InfrahubNode ) -> None :
3287 car = await client .get (kind = "TestCar" , name__value = "car_name" , prefetch_relationships = True )
88+ await car .delete ()
3389
34- owner_2 = await client .create (kind = "TestPerson" , name = " owner_2" )
35- await owner_2 .save ()
90+ owner_2 = await client .get (kind = "TestPerson" , name__value = owner_2 . name . value , prefetch_relationships = True )
91+ assert len ( owner_2 .cars . peers ) == 0
3692
37- car .owner = owner_2
38- await car .save ()
93+ async def test_delete_aware_node_with_agnostic_parent_blocked (
94+ self , client : InfrahubClient , car_2 : InfrahubNode , roofrack_2 : InfrahubNode
95+ ) -> None :
96+ car_2 = await client .get (kind = "TestCar" , name__value = car_2 .name .value , prefetch_relationships = True )
3997
40- car = await client .get (kind = "TestCar" , name__value = "car_name" , prefetch_relationships = True )
41- assert car .owner .peer .name .value == "owner_2"
98+ with pytest .raises (GraphQLError ) as exc :
99+ await car_2 .delete ()
100+
101+ assert (
102+ f"Cannot delete TestCar '{ car_2 .id } '. It is linked to mandatory relationship car on node TestRoofrack '{ roofrack_2 .id } '"
103+ in exc .value .message
104+ )
0 commit comments