Coverage for tests/config.py: 97%

1import logging 

2from uuid import uuid4 

3from dotenv import find_dotenv 

4from pydantic import AnyUrl, AnyHttpUrl, Field, AliasChoices, model_validator 

5from filip.models.base import FiwareHeader, LogLevel 

6from pydantic_settings import BaseSettings, SettingsConfigDict 

7from typing import Union, Optional 

8 

9 

10def generate_servicepath(): 

11 """ 

12 Generates a random service path 

13 

14 Returns: 

15 String: unique service path 

16 """ 

17 return f'/{str(uuid4()).replace("-", "")}' 

18 

19 

20class TestSettings(BaseSettings): 

21 """ 

22 Settings for the test case scenarios according to pydantic's documentaion 

23 https://pydantic-docs.helpmanual.io/usage/settings/ 

24 """ 

25 

26 LOG_LEVEL: LogLevel = Field( 

27 default=LogLevel.ERROR, validation_alias=AliasChoices("LOG_LEVEL", "LOGLEVEL") 

28 ) 

29 

30 CB_URL: AnyHttpUrl = Field( 

31 default="http://localhost:1026", 

32 validation_alias=AliasChoices( 

33 "ORION_URL", "CB_URL", "CB_HOST", "CONTEXTBROKER_URL", "OCB_URL" 

34 ), 

35 ) 

36 LD_CB_URL: AnyHttpUrl = Field( 

37 default="http://localhost:1026", 

38 validation_alias=AliasChoices( 

39 "LD_ORION_URL", "LD_CB_URL", "ORION_LD_URL", "SCORPIO_URL", "STELLIO_URL" 

40 ), 

41 ) 

42 IOTA_URL: AnyHttpUrl = Field( 

43 default="http://localhost:4041", validation_alias="IOTA_URL" 

44 ) 

45 IOTA_JSON_URL: AnyHttpUrl = Field( 

46 default="http://localhost:4041", validation_alias="IOTA_JSON_URL" 

47 ) 

48 

49 IOTA_UL_URL: AnyHttpUrl = Field( 

50 default="http://127.0.0.1:4061", validation_alias=AliasChoices("IOTA_UL_URL") 

51 ) 

52 

53 QL_URL: AnyHttpUrl = Field( 

54 default="http://127.0.0.1:8668", 

55 validation_alias=AliasChoices("QUANTUMLEAP_URL", "QL_URL"), 

56 ) 

57 

58 MQTT_BROKER_URL: AnyUrl = Field( 

59 default="mqtt://127.0.0.1:1883", 

60 validation_alias=AliasChoices("MQTT_BROKER_URL", "MQTT_URL", "MQTT_BROKER"), 

61 ) 

62 

63 MQTT_BROKER_URL_INTERNAL: AnyUrl = Field( 

64 default="mqtt://mosquitto:1883", 

65 validation_alias=AliasChoices("MQTT_BROKER_URL_INTERNAL", "MQTT_URL_INTERNAL"), 

66 ) 

67 

68 LD_MQTT_BROKER_URL: AnyUrl = Field( 

69 default="mqtt://127.0.0.1:1884", 

70 validation_alias=AliasChoices( 

71 "LD_MQTT_BROKER_URL", "LD_MQTT_URL", "LD_MQTT_BROKER" 

72 ), 

73 ) 

74 

75 LD_MQTT_BROKER_URL_INTERNAL: AnyUrl = Field( 

76 default="mqtt://mqtt-broker-ld:1883", 

77 validation_alias=AliasChoices( 

78 "LD_MQTT_BROKER_URL_INTERNAL", "LD_MQTT_URL_INTERNAL" 

79 ), 

80 ) 

81 

82 # IF CI_JOB_ID is present it will always overwrite the service path 

83 CI_JOB_ID: Optional[str] = Field( 

84 default=None, validation_alias=AliasChoices("CI_JOB_ID") 

85 ) 

86 

87 # create service paths for multi tenancy scenario and concurrent testing 

88 FIWARE_SERVICE: str = Field( 

89 default="filip", validation_alias=AliasChoices("FIWARE_SERVICE") 

90 ) 

91 

92 FIWARE_SERVICEPATH: str = Field( 

93 default_factory=generate_servicepath, 

94 validation_alias=AliasChoices( 

95 "FIWARE_PATH", "FIWARE_SERVICEPATH", "FIWARE_SERVICE_PATH" 

96 ), 

97 ) 

98 

99 STATIC_RECORDS: int = Field( 

100 default=1000, validation_alias=AliasChoices("STATIC_RECORDS") 

101 ) 

102 

103 @model_validator(mode="after") 

104 def generate_multi_tenancy_setup(cls, values): 

105 """ 

106 Tests if the fields for multi tenancy in fiware are consistent. 

107 If CI_JOB_ID is present it will always overwrite the service path. 

108 Args: 

109 values: class variables 

110 

111 Returns: 

112 

113 """ 

114 if values.model_dump().get("CI_JOB_ID", None): 

115 values.FIWARE_SERVICEPATH = f"/{values.CI_JOB_ID}" 

116 

117 # validate header 

118 FiwareHeader( 

119 service=values.FIWARE_SERVICE, service_path=values.FIWARE_SERVICEPATH 

120 ) 

121 

122 return values 

123 

124 model_config = SettingsConfigDict( 

125 env_file=find_dotenv(".env"), 

126 env_file_encoding="utf-8", 

127 case_sensitive=False, 

128 use_enum_values=True, 

129 ) 

130 

131 

132# create settings object 

133settings = TestSettings() 

134print( 

135 f"Running tests with the following settings: \n " 

136 f"{settings.model_dump_json(indent=2)}" 

137) 

138 

139# configure logging for all tests 

140logging.basicConfig( 

141 level=settings.LOG_LEVEL, format="%(asctime)s %(name)s %(levelname)s: %(message)s" 

142)