12211967770

Committed 07 Dec 2024 10:02AM CUT coverage: 89.481% (+0.03%) from 89.45%

Build # 12211967770

Build Type

Pull #146

github

Committed by

hugodutka

Commit Message

use double quotes instead of single quotes for parameter values on all platforms

Pull Request Pull Request #146: fix: StartParameters on Windows

Run Details

4 of 4 new or added lines in 1 file covered. (100.0%)

604 of 675 relevant lines covered (89.48%)

1218.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.36

/embedded_postgres.go

package embeddedpostgres

import (
        "errors"
        "fmt"
        "net"
        "os"
        "os/exec"
        "path/filepath"
        "runtime"
        "strings"
        "sync"
)

var mu sync.Mutex

var (
        ErrServerNotStarted     = errors.New("server has not been started")
        ErrServerAlreadyStarted = errors.New("server is already started")
)

// EmbeddedPostgres maintains all configuration and runtime functions for maintaining the lifecycle of one Postgres process.
type EmbeddedPostgres struct {
        config              Config
        cacheLocator        CacheLocator
        remoteFetchStrategy RemoteFetchStrategy
        initDatabase        initDatabase
        createDatabase      createDatabase
        started             bool
        syncedLogger        *syncedLogger
}

// NewDatabase creates a new EmbeddedPostgres struct that can be used to start and stop a Postgres process.
// When called with no parameters it will assume a default configuration state provided by the DefaultConfig method.
// When called with parameters the first Config parameter will be used for configuration.
func NewDatabase(config ...Config) *EmbeddedPostgres {
        if len(config) < 1 {
                return newDatabaseWithConfig(DefaultConfig())
        }

        return newDatabaseWithConfig(config[0])
}

func newDatabaseWithConfig(config Config) *EmbeddedPostgres {
        versionStrategy := defaultVersionStrategy(
                config,
                runtime.GOOS,
                runtime.GOARCH,
                linuxMachineName,
                shouldUseAlpineLinuxBuild,
        )
        cacheLocator := defaultCacheLocator(config.cachePath, versionStrategy)
        remoteFetchStrategy := defaultRemoteFetchStrategy(config.binaryRepositoryURL, versionStrategy, cacheLocator)

        return &EmbeddedPostgres{
                config:              config,
                cacheLocator:        cacheLocator,
                remoteFetchStrategy: remoteFetchStrategy,
                initDatabase:        defaultInitDatabase,
                createDatabase:      defaultCreateDatabase,
                started:             false,
        }
}

// Start will try to start the configured Postgres process returning an error when there were any problems with invocation.
// If any error occurs Start will try to also Stop the Postgres process in order to not leave any sub-process running.
//
//nolint:funlen
func (ep *EmbeddedPostgres) Start() error {
        if ep.started {
                return ErrServerAlreadyStarted
        }

        if err := ensurePortAvailable(ep.config.port); err != nil {
                return err
        }

        logger, err := newSyncedLogger("", ep.config.logger)
        if err != nil {
                return errors.New("unable to create logger")
        }

        ep.syncedLogger = logger

        cacheLocation, cacheExists := ep.cacheLocator()

        if ep.config.runtimePath == "" {
                ep.config.runtimePath = filepath.Join(filepath.Dir(cacheLocation), "extracted")
        }

        if ep.config.dataPath == "" {
                ep.config.dataPath = filepath.Join(ep.config.runtimePath, "data")
        }

        if err := os.RemoveAll(ep.config.runtimePath); err != nil {
                return fmt.Errorf("unable to clean up runtime directory %s with error: %s", ep.config.runtimePath, err)
        }

        if ep.config.binariesPath == "" {
                ep.config.binariesPath = ep.config.runtimePath
        }

        if err := ep.downloadAndExtractBinary(cacheExists, cacheLocation); err != nil {
                return err
        }

        if err := os.MkdirAll(ep.config.runtimePath, os.ModePerm); err != nil {
                return fmt.Errorf("unable to create runtime directory %s with error: %s", ep.config.runtimePath, err)
        }

        reuseData := dataDirIsValid(ep.config.dataPath, ep.config.version)

        if !reuseData {
                if err := ep.cleanDataDirectoryAndInit(); err != nil {
                        return err
                }
        }

        if err := startPostgres(ep); err != nil {
                return err
        }

        if err := ep.syncedLogger.flush(); err != nil {
                return err
        }

        ep.started = true

        if !reuseData {
                if err := ep.createDatabase(ep.config.port, ep.config.username, ep.config.password, ep.config.database); err != nil {
                        if stopErr := stopPostgres(ep); stopErr != nil {
                                return fmt.Errorf("unable to stop database caused by error %s", err)
                        }

                        return err
                }
        }

        if err := healthCheckDatabaseOrTimeout(ep.config); err != nil {
                if stopErr := stopPostgres(ep); stopErr != nil {
                        return fmt.Errorf("unable to stop database caused by error %s", err)
                }

                return err
        }

        return nil
}

func (ep *EmbeddedPostgres) downloadAndExtractBinary(cacheExists bool, cacheLocation string) error {
        // lock to prevent collisions with duplicate downloads
        mu.Lock()
        defer mu.Unlock()

        _, binDirErr := os.Stat(filepath.Join(ep.config.binariesPath, "bin", "pg_ctl"))
        if os.IsNotExist(binDirErr) {
                if !cacheExists {
                        if err := ep.remoteFetchStrategy(); err != nil {
                                return err
                        }
                }

                if err := decompressTarXz(defaultTarReader, cacheLocation, ep.config.binariesPath); err != nil {
                        return err
                }
        }
        return nil
}

func (ep *EmbeddedPostgres) cleanDataDirectoryAndInit() error {
        if err := os.RemoveAll(ep.config.dataPath); err != nil {
                return fmt.Errorf("unable to clean up data directory %s with error: %s", ep.config.dataPath, err)
        }

        if err := ep.initDatabase(ep.config.binariesPath, ep.config.runtimePath, ep.config.dataPath, ep.config.username, ep.config.password, ep.config.locale, ep.config.encoding, ep.syncedLogger.file); err != nil {
                return err
        }

        return nil
}

// Stop will try to stop the Postgres process gracefully returning an error when there were any problems.
func (ep *EmbeddedPostgres) Stop() error {
        if !ep.started {
                return ErrServerNotStarted
        }

        if err := stopPostgres(ep); err != nil {
                return err
        }

        ep.started = false

        if err := ep.syncedLogger.flush(); err != nil {
                return err
        }

        return nil
}

func encodeOptions(port uint32, parameters map[string]string) string {
        options := []string{fmt.Sprintf("-p %d", port)}
        for k, v := range parameters {
                // Double-quote parameter values - they may have spaces.
                // Careful: CMD on Windows uses only double quotes to delimit strings.
                // It treats single quotes as regular characters.
                options = append(options, fmt.Sprintf("-c %s=\"%s\"", k, v))
        }
        return strings.Join(options, " ")
}

func startPostgres(ep *EmbeddedPostgres) error {
        postgresBinary := filepath.Join(ep.config.binariesPath, "bin/pg_ctl")
        postgresProcess := exec.Command(postgresBinary, "start", "-w",
                "-D", ep.config.dataPath,
                "-o", encodeOptions(ep.config.port, ep.config.startParameters))
        postgresProcess.Stdout = ep.syncedLogger.file
        postgresProcess.Stderr = ep.syncedLogger.file

        if err := postgresProcess.Run(); err != nil {
                _ = ep.syncedLogger.flush()
                logContent, _ := readLogsOrTimeout(ep.syncedLogger.file)

                return fmt.Errorf("could not start postgres using %s:\n%s", postgresProcess.String(), string(logContent))
        }

        return nil
}

func stopPostgres(ep *EmbeddedPostgres) error {
        postgresBinary := filepath.Join(ep.config.binariesPath, "bin/pg_ctl")
        postgresProcess := exec.Command(postgresBinary, "stop", "-w",
                "-D", ep.config.dataPath)
        postgresProcess.Stderr = ep.syncedLogger.file
        postgresProcess.Stdout = ep.syncedLogger.file

        if err := postgresProcess.Run(); err != nil {
                return err
        }

        return nil
}

func ensurePortAvailable(port uint32) error {
        conn, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", port))
        if err != nil {
                return fmt.Errorf("process already listening on port %d", port)
        }

        if err := conn.Close(); err != nil {
                return err
        }

        return nil
}

func dataDirIsValid(dataDir string, version PostgresVersion) bool {
        pgVersion := filepath.Join(dataDir, "PG_VERSION")

        d, err := os.ReadFile(pgVersion)
        if err != nil {
                return false
        }

        v := strings.TrimSuffix(string(d), "\n")

        return strings.HasPrefix(string(version), v)
}

1	package embeddedpostgres
2
3	import (
4	"errors"
5	"fmt"
6	"net"
7	"os"
8	"os/exec"
9	"path/filepath"
10	"runtime"
11	"strings"
12	"sync"
13	)
14
15	var mu sync.Mutex
16
17	var (
18	ErrServerNotStarted = errors.New("server has not been started")
19	ErrServerAlreadyStarted = errors.New("server is already started")
20	)
21
22	// EmbeddedPostgres maintains all configuration and runtime functions for maintaining the lifecycle of one Postgres process.
23	type EmbeddedPostgres struct {
24	config Config
25	cacheLocator CacheLocator
26	remoteFetchStrategy RemoteFetchStrategy
27	initDatabase initDatabase
28	createDatabase createDatabase
29	started bool
30	syncedLogger *syncedLogger
31	}
32
33	// NewDatabase creates a new EmbeddedPostgres struct that can be used to start and stop a Postgres process.
34	// When called with no parameters it will assume a default configuration state provided by the DefaultConfig method.
35	// When called with parameters the first Config parameter will be used for configuration.
36	func NewDatabase(config ...Config) *EmbeddedPostgres {	31✔
37	if len(config) < 1 {	37✔
38	return newDatabaseWithConfig(DefaultConfig())	6✔
39	}	6✔
40
41	return newDatabaseWithConfig(config[0])	25✔
42	}
43
44	func newDatabaseWithConfig(config Config) *EmbeddedPostgres {	31✔
45	versionStrategy := defaultVersionStrategy(	31✔
46	config,	31✔
47	runtime.GOOS,	31✔
48	runtime.GOARCH,	31✔
49	linuxMachineName,	31✔
50	shouldUseAlpineLinuxBuild,	31✔
51	)	31✔
52	cacheLocator := defaultCacheLocator(config.cachePath, versionStrategy)	31✔
53	remoteFetchStrategy := defaultRemoteFetchStrategy(config.binaryRepositoryURL, versionStrategy, cacheLocator)	31✔
54		31✔
55	return &EmbeddedPostgres{	31✔
56	config: config,	31✔
57	cacheLocator: cacheLocator,	31✔
58	remoteFetchStrategy: remoteFetchStrategy,	31✔
59	initDatabase: defaultInitDatabase,	31✔
60	createDatabase: defaultCreateDatabase,	31✔
61	started: false,	31✔
62	}	31✔
63	}	31✔
64
65	// Start will try to start the configured Postgres process returning an error when there were any problems with invocation.
66	// If any error occurs Start will try to also Stop the Postgres process in order to not leave any sub-process running.
67	//
68	//nolint:funlen
69	func (ep *EmbeddedPostgres) Start() error {	32✔
70	if ep.started {	33✔
71	return ErrServerAlreadyStarted	1✔
72	}	1✔
73
74	if err := ensurePortAvailable(ep.config.port); err != nil {	32✔
75	return err	1✔
76	}	1✔
77
78	logger, err := newSyncedLogger("", ep.config.logger)	30✔
79	if err != nil {	30✔
80	return errors.New("unable to create logger")	×
81	}	×
82
83	ep.syncedLogger = logger	30✔
84		30✔
85	cacheLocation, cacheExists := ep.cacheLocator()	30✔
86		30✔
87	if ep.config.runtimePath == "" {	46✔
88	ep.config.runtimePath = filepath.Join(filepath.Dir(cacheLocation), "extracted")	16✔
89	}	16✔
90
91	if ep.config.dataPath == "" {	56✔
92	ep.config.dataPath = filepath.Join(ep.config.runtimePath, "data")	26✔
93	}	26✔
94
95	if err := os.RemoveAll(ep.config.runtimePath); err != nil {	30✔
96	return fmt.Errorf("unable to clean up runtime directory %s with error: %s", ep.config.runtimePath, err)	×
97	}	×
98
99	if ep.config.binariesPath == "" {	56✔
100	ep.config.binariesPath = ep.config.runtimePath	26✔
101	}	26✔
102
103	if err := ep.downloadAndExtractBinary(cacheExists, cacheLocation); err != nil {	32✔
104	return err	2✔
105	}	2✔
106
107	if err := os.MkdirAll(ep.config.runtimePath, os.ModePerm); err != nil {	28✔
108	return fmt.Errorf("unable to create runtime directory %s with error: %s", ep.config.runtimePath, err)	×
109	}	×
110
111	reuseData := dataDirIsValid(ep.config.dataPath, ep.config.version)	28✔
112		28✔
113	if !reuseData {	55✔
114	if err := ep.cleanDataDirectoryAndInit(); err != nil {	28✔
115	return err	1✔
116	}	1✔
117	}
118
119	if err := startPostgres(ep); err != nil {	28✔
120	return err	1✔
121	}	1✔
122
123	if err := ep.syncedLogger.flush(); err != nil {	26✔
124	return err	×
125	}	×
126
127	ep.started = true	26✔
128		26✔
129	if !reuseData {	51✔
130	if err := ep.createDatabase(ep.config.port, ep.config.username, ep.config.password, ep.config.database); err != nil {	26✔
131	if stopErr := stopPostgres(ep); stopErr != nil {	1✔
132	return fmt.Errorf("unable to stop database caused by error %s", err)	×
133	}	×
134
135	return err	1✔
136	}
137	}
138
139	if err := healthCheckDatabaseOrTimeout(ep.config); err != nil {	26✔
140	if stopErr := stopPostgres(ep); stopErr != nil {	1✔
141	return fmt.Errorf("unable to stop database caused by error %s", err)	×
142	}	×
143
144	return err	1✔
145	}
146
147	return nil	24✔
148	}
149
150	func (ep *EmbeddedPostgres) downloadAndExtractBinary(cacheExists bool, cacheLocation string) error {	30✔
151	// lock to prevent collisions with duplicate downloads	30✔
152	mu.Lock()	30✔
153	defer mu.Unlock()	30✔
154		30✔
155	_, binDirErr := os.Stat(filepath.Join(ep.config.binariesPath, "bin", "pg_ctl"))	30✔
156	if os.IsNotExist(binDirErr) {	58✔
157	if !cacheExists {	35✔
158	if err := ep.remoteFetchStrategy(); err != nil {	8✔
159	return err	1✔
160	}	1✔
161	}
162
163	if err := decompressTarXz(defaultTarReader, cacheLocation, ep.config.binariesPath); err != nil {	28✔
164	return err	1✔
165	}	1✔
166	}
167	return nil	28✔
168	}
169
170	func (ep *EmbeddedPostgres) cleanDataDirectoryAndInit() error {	27✔
171	if err := os.RemoveAll(ep.config.dataPath); err != nil {	27✔
172	return fmt.Errorf("unable to clean up data directory %s with error: %s", ep.config.dataPath, err)	×
173	}	×
174
175	if err := ep.initDatabase(ep.config.binariesPath, ep.config.runtimePath, ep.config.dataPath, ep.config.username, ep.config.password, ep.config.locale, ep.config.encoding, ep.syncedLogger.file); err != nil {	28✔
176	return err	1✔
177	}	1✔
178
179	return nil	26✔
180	}
181
182	// Stop will try to stop the Postgres process gracefully returning an error when there were any problems.
183	func (ep *EmbeddedPostgres) Stop() error {	25✔
184	if !ep.started {	26✔
185	return ErrServerNotStarted	1✔
186	}	1✔
187
188	if err := stopPostgres(ep); err != nil {	24✔
189	return err	×
190	}	×
191
192	ep.started = false	24✔
193		24✔
194	if err := ep.syncedLogger.flush(); err != nil {	24✔
195	return err	×
196	}	×
197
198	return nil	24✔
199	}
200
201	func encodeOptions(port uint32, parameters map[string]string) string {	27✔
202	options := []string{fmt.Sprintf("-p %d", port)}	27✔
203	for k, v := range parameters {	28✔
204	// Double-quote parameter values - they may have spaces.	1✔
205	// Careful: CMD on Windows uses only double quotes to delimit strings.	1✔
206	// It treats single quotes as regular characters.	1✔
207	options = append(options, fmt.Sprintf("-c %s=\"%s\"", k, v))	1✔
208	}	1✔
209	return strings.Join(options, " ")	27✔
210	}
211
212	func startPostgres(ep *EmbeddedPostgres) error {	27✔
213	postgresBinary := filepath.Join(ep.config.binariesPath, "bin/pg_ctl")	27✔
214	postgresProcess := exec.Command(postgresBinary, "start", "-w",	27✔
215	"-D", ep.config.dataPath,	27✔
216	"-o", encodeOptions(ep.config.port, ep.config.startParameters))	27✔
217	postgresProcess.Stdout = ep.syncedLogger.file	27✔
218	postgresProcess.Stderr = ep.syncedLogger.file	27✔
219		27✔
220	if err := postgresProcess.Run(); err != nil {	28✔
221	_ = ep.syncedLogger.flush()	1✔
222	logContent, _ := readLogsOrTimeout(ep.syncedLogger.file)	1✔
223		1✔
224	return fmt.Errorf("could not start postgres using %s:\n%s", postgresProcess.String(), string(logContent))	1✔
225	}	1✔
226
227	return nil	26✔
228	}
229
230	func stopPostgres(ep *EmbeddedPostgres) error {	26✔
231	postgresBinary := filepath.Join(ep.config.binariesPath, "bin/pg_ctl")	26✔
232	postgresProcess := exec.Command(postgresBinary, "stop", "-w",	26✔
233	"-D", ep.config.dataPath)	26✔
234	postgresProcess.Stderr = ep.syncedLogger.file	26✔
235	postgresProcess.Stdout = ep.syncedLogger.file	26✔
236		26✔
237	if err := postgresProcess.Run(); err != nil {	26✔
238	return err	×
239	}	×
240
241	return nil	26✔
242	}
243
244	func ensurePortAvailable(port uint32) error {	31✔
245	conn, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", port))	31✔
246	if err != nil {	32✔
247	return fmt.Errorf("process already listening on port %d", port)	1✔
248	}	1✔
249
250	if err := conn.Close(); err != nil {	30✔
251	return err	×
252	}	×
253
254	return nil	30✔
255	}
256
257	func dataDirIsValid(dataDir string, version PostgresVersion) bool {	28✔
258	pgVersion := filepath.Join(dataDir, "PG_VERSION")	28✔
259		28✔
260	d, err := os.ReadFile(pgVersion)	28✔
261	if err != nil {	55✔
262	return false	27✔
263	}	27✔
264
265	v := strings.TrimSuffix(string(d), "\n")	1✔
266		1✔
267	return strings.HasPrefix(string(version), v)	1✔
268	}

fergusstrange / embedded-postgres / 12211967770

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous