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

48 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-24 12:46 +0200

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 

77def create_tables(): 

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

79 

80 u.pg.create_schema(SCHEMA) 

81 

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

83 'id': 'int primary key', 

84 'name': 'text', 

85 'district_id': 'int', 

86 'photo': 'text', 

87 'photo_content': 'bytea', 

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

89 }) 

90 

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

92 'id': 'int primary key', 

93 'name': 'text', 

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

95 }) 

96 

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

98 'id': 'serial primary key', 

99 'fid': 'int', 

100 'kind': 'text', 

101 'text': 'text', 

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

103 }) 

104 

105 

106def point(x, y): 

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

108 

109 

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

111 return u.pg.ewkb( 

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

113 3857, 

114 )