1
1
import os
2
2
from typing import Optional
3
3
from testcontainers .core .generic import DbContainer
4
+
4
5
# from testcontainers.core.utils import raise_for_deprecated_parameter
5
6
from testcontainers .core .waiting_utils import wait_for_logs
6
7
8
+
7
9
class IRISContainer (DbContainer ):
8
10
"""
9
11
InterSystems IRIS database container.
@@ -27,15 +29,18 @@ class IRISContainer(DbContainer):
27
29
>>> version
28
30
'IRIS for UNIX (Ubuntu Server LTS for ARM64 Containers) 2023.2 (Build 227U) Mon Jul 31 2023 17:43:25 EDT'
29
31
"""
30
- def __init__ (self , image : str = "intersystemsdc/iris-community:latest" ,
31
- port : int = 1972 ,
32
- username : Optional [str ] = None ,
33
- password : Optional [str ] = None ,
34
- namespace : Optional [str ] = None ,
35
- driver : str = "iris" ,
36
- license_key : str = None ,
37
- ** kwargs
38
- ) -> None :
32
+
33
+ def __init__ (
34
+ self ,
35
+ image : str = "intersystemsdc/iris-community:latest" ,
36
+ port : int = 1972 ,
37
+ username : Optional [str ] = None ,
38
+ password : Optional [str ] = None ,
39
+ namespace : Optional [str ] = None ,
40
+ driver : str = "iris" ,
41
+ license_key : str = None ,
42
+ ** kwargs ,
43
+ ) -> None :
39
44
# raise_for_deprecated_parameter(kwargs, "user", "username")
40
45
super (IRISContainer , self ).__init__ (image = image , ** kwargs )
41
46
self .image = image
@@ -53,28 +58,39 @@ def _configure(self) -> None:
53
58
self .with_env ("IRIS_PASSWORD" , self .password )
54
59
self .with_env ("IRIS_NAMESPACE" , self .namespace )
55
60
if self .license_key :
56
- self .with_volume_mapping (self .license_key , "/usr/irissys/mgr/iris.key" , "ro" )
61
+ self .with_volume_mapping (
62
+ self .license_key , "/usr/irissys/mgr/iris.key" , "ro"
63
+ )
64
+
57
65
def _connect (self ) -> None :
58
66
wait_for_logs (self , predicate = "Enabling logons" )
59
67
if self .image .startswith ("intersystemsdc" ):
60
68
wait_for_logs (self , predicate = "executed command" )
61
69
else :
62
70
if self .namespace .upper () != "USER" :
63
- cmd = f"iris session iris -U %%SYS '##class(%%SQL.Statement).%%ExecDirect(,\" CREATE DATABASE %s\" )'" % (
64
- self .namespace ,
71
+ cmd = (
72
+ f"iris session iris -U %%SYS '##class(%%SQL.Statement).%%ExecDirect(,\" CREATE DATABASE %s\" )'"
73
+ % (self .namespace ,)
65
74
)
66
75
self .exec (cmd )
67
- cmd = f"iris session iris -U %%SYS '##class(Security.Users).Create(\" %s\" ,\" %s\" ,\" %s\" ,,\" %s\" )'" % (
68
- self .username ,
69
- "%ALL" ,
70
- self .password ,
71
- self .namespace ,
76
+ cmd = (
77
+ f'iris session iris -U %%SYS \' ##class(Security.Users).Create("%s","%s","%s",,"%s")\' '
78
+ % (
79
+ self .username ,
80
+ "%ALL" ,
81
+ self .password ,
82
+ self .namespace ,
83
+ )
72
84
)
73
85
res = self .exec (cmd )
86
+ print ("res" , cmd , res )
74
87
75
88
def get_connection_url (self , host = None ) -> str :
76
89
return super ()._create_connection_url (
77
- dialect = f"iris" , username = self .username ,
78
- password = self .password , db_name = self .namespace , host = host ,
90
+ dialect = "iris" ,
91
+ username = self .username ,
92
+ password = self .password ,
93
+ db_name = self .namespace ,
94
+ host = host ,
79
95
port = self .port ,
80
96
)
0 commit comments