def _ensure_repo()

in src/buildstream_plugins/sources/git.py [0:0]

• 17 lines of code
• 4 McCabe index (conditional complexity)

    def _ensure_repo(self):
        if not os.path.exists(self.mirror):
            with self.source.tempdir() as tmpdir:
                self.source.call(
                    [self.source.host_git, "init", "--bare", tmpdir],
                    fail="Failed to initialise repository",
                )

                try:
                    utils.move_atomic(tmpdir, self.mirror)
                except DirectoryExistsError:
                    # Another process was quicker to download this repository.
                    # Let's discard our own
                    self.source.status("{}: Discarding duplicate repository".format(self.source))
                except OSError as e:
                    raise SourceError(
                        "{}: Failed to move created repository from '{}' to '{}': {}".format(
                            self.source, tmpdir, self.mirror, e
                        )
                    ) from e