Coverage for gws-app/gws/plugin/qfieldcloud/_test/util.py: 95%

1"""Shared fixtures for qfieldcloud tests.""" 

2 

3import os 

4 

5import gws 

6import gws.lib.xmlx 

7import gws.test.util as u 

8 

9SCHEMA = 'qfc' 

10 

11SOURCE_QGS = os.path.dirname(__file__) + '/qfield.qgs' 

12 

13# the qgis container mounts VAR_DIR under the same path, so projects placed here 

14# are readable by the qgis server as well 

15WORK_DIR = f'{gws.c.VAR_DIR}/qfieldcloud_test' 

16 

17 

18def qgs_path(name: str = 'qfield', patch=None) -> str: 

19 """Copy the fixture qgis project into the shared work dir and return its path. 

20 

21 Args: 

22 name: Base name for the copy. 

23 patch: Optional callable, receives the parsed xml root before writing. 

24 """ 

25 

26 text = gws.u.read_file(SOURCE_QGS) 

27 text = text.replace('{PG_SERVICE}', u.option('service.postgres.name')) 

28 

29 if patch: 

30 root_el = gws.lib.xmlx.from_string(text) 

31 patch(root_el) 

32 text = root_el.to_string() 

33 

34 path = f'{gws.u.ensure_dir(WORK_DIR)}/{name}.qgs' 

35 gws.u.write_file(path, text) 

36 return path 

37 

38 

39def set_project_prop(root_el: gws.XmlElement, name: str, value: str, type: str = 'QString'): 

40 """Set a qfieldsync project property, for use as a `qgs_path` patch.""" 

41 

42 el = root_el.require('properties/qfieldsync') 

43 old = el.find(name) 

44 if old is not None: 

45 el.remove(old) 

46 el.add(name, {'type': type}).text = value 

47 

48 

49def remove_media_dirs(root_el: gws.XmlElement): 

50 """Drop the media dirs, for use as a `qgs_path` patch. 

51 

52 All fixture projects live in the same directory, so a test that creates media files 

53 would otherwise affect every other project as well. 

54 """ 

55 

56 el = root_el.require('properties') 

57 old = el.find('QFieldSync') 

58 if old is not None: 

59 el.remove(old) 

60 

61 

62def set_layer_prop(root_el: gws.XmlElement, layer_id: str, name: str, value: str, type: str = 'QString'): 

63 """Set a QFieldSync layer custom property, for use as a `qgs_path` patch.""" 

64 

65 for el in root_el.findall('projectlayers/maplayer'): 

66 if el.textof('id') != layer_id: 

67 continue 

68 opt = el.require('customproperties/Option') 

69 for o in opt.findall('Option'): 

70 if o.get('name') == f'QFieldSync/{name}': 

71 opt.remove(o) 

72 opt.add('Option', {'name': f'QFieldSync/{name}', 'type': type, 'value': value}) 

73 return 

74 raise ValueError(f'layer {layer_id!r} not found') 

75 

76 

77DB_UID = 'QFC_DB' 

78 

79 

80def db_provider_config() -> str: 

81 """Return the config of a database provider with the same connection as the fixture QGIS project. 

82 

83 Models configured with ``dbUid DB_UID`` and the models created for the QGIS layers then share one provider. 

84 """ 

85 

86 name = u.option('service.postgres.name') 

87 return f'database.providers+ {{ uid "{DB_UID}" type "postgres" serviceName "{name}" schemaCacheLifeTime 0 }}' 

88 

89 

90def remove_deltas(root: gws.Root, action_uid: str = 'ACTION_1'): 

91 """Remove the delta databases of the QField projects of an action.""" 

92 

93 act = root.get(action_uid) 

94 for qp in act.qfcProjects: 

95 path = act.deltas_db_path(qp) 

96 if os.path.exists(path): 

97 os.unlink(path) 

98 

99 

100def create_tables(): 

101 """Create the postgres tables the fixture project refers to.""" 

102 

103 u.pg.create_schema(SCHEMA) 

104 

105 u.pg.create(f'{SCHEMA}.poi', { 

106 'id': 'int primary key', 

107 'name': 'text', 

108 'district_id': 'int', 

109 'photo': 'text', 

110 'photo_content': 'bytea', 

111 'geom': 'geometry(Point,3857)', 

112 }) 

113 

114 u.pg.create(f'{SCHEMA}.district', { 

115 'id': 'int primary key', 

116 'name': 'text', 

117 'geom': 'geometry(MultiPolygon,3857)', 

118 }) 

119 

120 u.pg.create(f'{SCHEMA}.note', { 

121 'id': 'serial primary key', 

122 'fid': 'int', 

123 'kind': 'text', 

124 'text': 'text', 

125 'geom': 'geometry(Point,3857)', 

126 }) 

127 

128 

129def point(x, y): 

130 return u.pg.ewkb(f'POINT({x} {y})', 3857) 

131 

132 

133def polygon(x, y, size=1000): 

134 return u.pg.ewkb( 

135 f'MULTIPOLYGON((({x} {y}, {x + size} {y}, {x + size} {y + size}, {x} {y + size}, {x} {y})))', 

136 3857, 

137 )