YQ Connector:Use docker-compose in integrational tests

5 files changed, 520 insertions, 0 deletions

diff --git a/library/cpp/zipatch/reader.cpp b/library/cpp/zipatch/reader.cpp
new file mode 100644
index 0000000000..03ac365da1
--- /dev/null
+++ b/library/cpp/zipatch/reader.cpp
@@ -0,0 +1,173 @@
+#include "reader.h"
+
+#include <library/cpp/json/json_reader.h>
+#include <library/cpp/json/json_value.h>
+
+#include <util/generic/hash.h>
+#include <util/memory/tempbuf.h>
+
+#include <contrib/libs/libarchive/libarchive/archive.h>
+#include <contrib/libs/libarchive/libarchive/archive_entry.h>
+
+using namespace NJson;
+
+namespace NZipatch {
+
+class TReader::TImpl {
+
+    using TEntry = archive_entry;
+
+public:
+    TImpl() {
+        if ((Archive_ = archive_read_new()) == nullptr) {
+            ythrow yexception() << "can't create archive object";
+        }
+    }
+
+    TImpl(const TFsPath& path)
+        : TImpl()
+    {
+        archive_read_support_filter_all(Archive_);
+        archive_read_support_format_zip(Archive_);
+
+        if (ARCHIVE_OK != archive_read_open_filename(Archive_, TString(path).c_str(), 10240)) {
+            ythrow yexception() << "can't open archive path = " << path;
+        }
+
+        Read();
+    }
+
+    TImpl(const TStringBuf buf)
+        : TImpl()
+    {
+        archive_read_support_filter_all(Archive_);
+        archive_read_support_format_zip(Archive_);
+
+        if (ARCHIVE_OK != archive_read_open_memory(Archive_, buf.data(), buf.size())) {
+            ythrow yexception() << "can't open in-memory archive";
+        }
+
+        Read();
+    }
+
+    ~TImpl() {
+        for (const auto& item : Files_) {
+            archive_entry_free(item.second.first);
+        }
+        if (Archive_) {
+            archive_read_free(Archive_);
+        }
+    }
+
+    void Enumerate(TOnEvent cb) const {
+        for (const auto& item : Actions_) {
+            TEvent event;
+
+            event.Action = GetTypeFromString(item["type"].GetStringSafe(TString()));
+            event.Path = item["path"].GetStringSafe(TString());
+            event.Executable = item["executable"].GetBooleanSafe(false);
+            event.Symlink = false;
+
+            if (event.Action == Copy || event.Action == Move) {
+                event.Source.Path = item["orig_path"].GetStringSafe(TString());
+                event.Source.Revision = item["orig_revision"].GetUIntegerRobust();
+            }
+            if (event.Action == StoreFile) {
+                auto fi = Files_.find(event.Path);
+                if (fi == Files_.end()) {
+                    ythrow yexception() << "can't find file; path = " << event.Path;
+                }
+
+                event.Data = fi->second.second;
+                event.Symlink = archive_entry_filetype(fi->second.first) == AE_IFLNK;
+            }
+
+            if (event.Path) {
+                cb(event);
+            }
+        }
+    }
+
+private:
+    EAction GetTypeFromString(const TString& type) const {
+        if (type == "store_file") {
+            return StoreFile;
+        }
+        if (type == "mkdir") {
+            return MkDir;
+        }
+        if (type == "remove_file" || type == "remove_tree") {
+            return Remove;
+        }
+        if (type == "svn_copy") {
+            return Copy;
+        }
+        return Unknown;
+    }
+
+    void Read() {
+        TEntry* current = nullptr;
+
+        while (archive_read_next_header(Archive_, &current) == ARCHIVE_OK) {
+            const TStringBuf path(archive_entry_pathname(current));
+
+            if (path == "actions.json") {
+                TJsonValue value;
+                ReadJsonFastTree(GetData(current), &value, true);
+
+                for (const auto& item : value.GetArraySafe()) {
+                    Actions_.push_back(item);
+                }
+            } else if (AsciiHasPrefix(path, "files/")) {
+                TEntry* entry = archive_entry_clone(current);
+
+                Files_.emplace(path.substr(6), std::make_pair(entry, GetData(current)));
+            }
+        }
+
+        archive_read_close(Archive_);
+    }
+
+    TString GetData(TEntry* current) const {
+        if (archive_entry_filetype(current) == AE_IFLNK) {
+            return archive_entry_symlink(current);
+        }
+
+        if (const auto size = archive_entry_size(current)) {
+            TTempBuf data(size);
+
+            if (archive_read_data(Archive_, data.Data(), size) != size) {
+                ythrow yexception() << "can't read entry";
+            }
+
+            return TString(data.Data(), size);
+        }
+
+        return TString();
+    }
+
+private:
+    struct archive* Archive_;
+    TVector<TJsonValue> Actions_;
+    THashMap<TString, std::pair<TEntry*, TString>> Files_;
+};
+
+TReader::TReader(const TFsPath& path)
+    : Impl_(new TImpl(path))
+{
+}
+
+TReader::TReader(const TStringBuf buf)
+    : Impl_(new TImpl(buf))
+{
+}
+
+TReader::~TReader()
+{ }
+
+void TReader::Enumerate(TOnEvent cb) const {
+    Impl_->Enumerate(cb);
+}
+
+} // namespace NZipatch
+
diff --git a/library/cpp/zipatch/reader.h b/library/cpp/zipatch/reader.h
new file mode 100644
index 0000000000..a94bc79b71
--- /dev/null
+++ b/library/cpp/zipatch/reader.h
@@ -0,0 +1,48 @@
+#pragma once
+
+#include <util/folder/path.h>
+#include <util/generic/ptr.h>
+
+namespace NZipatch {
+
+class TReader {
+public:
+    enum EAction {
+        Unknown = 0,
+        Copy,
+        MkDir,
+        Move,
+        Remove,
+        StoreFile,
+    };
+
+    struct TSource {
+        TString Path;
+        ui64 Revision;
+    };
+
+    struct TEvent {
+        EAction Action;
+        TString Path;
+        TStringBuf Data;
+        TSource Source;
+        bool Executable;
+        bool Symlink;
+    };
+
+    using TOnEvent = std::function<void(const TEvent&)>;
+
+public:
+     TReader(const TFsPath& path);
+     TReader(const TStringBuf buf);
+    ~TReader();
+
+    void Enumerate(TOnEvent cb) const;
+
+private:
+    class TImpl;
+    THolder<TImpl> Impl_;
+};
+
+} // namespace NZipatch
+
diff --git a/library/cpp/zipatch/writer.cpp b/library/cpp/zipatch/writer.cpp
new file mode 100644
index 0000000000..a9ca451b01
--- /dev/null
+++ b/library/cpp/zipatch/writer.cpp
@@ -0,0 +1,232 @@
+#include "writer.h"
+
+#include <library/cpp/json/json_value.h>
+#include <library/cpp/json/json_writer.h>
+
+#include <util/string/join.h>
+
+#include <contrib/libs/libarchive/libarchive/archive.h>
+#include <contrib/libs/libarchive/libarchive/archive_entry.h>
+
+using namespace NJson;
+
+namespace NZipatch {
+
+class TWriter::TImpl {
+public:
+    TImpl(const TFsPath& path)
+        : Actions_(new TJsonValue(JSON_ARRAY))
+        , Meta_(new TJsonValue(JSON_MAP))
+        , Revprops_(new TJsonValue(JSON_MAP))
+        , Archive_(nullptr)
+    {
+        Archive_ = archive_write_new();
+        if (!Archive_) {
+            ythrow yexception() << "can't create archive object";
+        }
+        archive_write_set_format_zip(Archive_);
+        archive_write_zip_set_compression_deflate(Archive_);
+
+        if (ARCHIVE_OK != archive_write_open_filename(Archive_, TString(path).c_str())) {
+            ythrow yexception() << "can't open archive path = " << path;
+        }
+    }
+
+    ~TImpl() {
+        if (Actions_ || Meta_ || Revprops_) {
+            Finish();
+        }
+        if (Archive_) {
+            archive_write_free(Archive_);
+        }
+    }
+
+    void Finish() {
+        if (Actions_) {
+            if (Archive_) {
+                WriteEntry("actions.json", WriteJson(Actions_.Get(), true, false));
+            }
+
+            Actions_.Destroy();
+        }
+
+        if (Meta_) {
+            if (Archive_) {
+                WriteEntry("meta.json", WriteJson(Meta_.Get(), true));
+            }
+
+            Meta_.Destroy();
+        }
+
+        if (Revprops_) {
+            if (Archive_) {
+                WriteEntry("revprops.json", WriteJson(Revprops_.Get(), true));
+            }
+
+            Revprops_.Destroy();
+        }
+
+        if (Archive_) {
+            archive_write_close(Archive_);
+        }
+    }
+
+    void Copy(const TString& path, const TOrigin& origin) {
+        Y_ASSERT(origin.Path);
+        Y_ASSERT(origin.Revision);
+
+        if (Actions_) {
+            TJsonValue item;
+            item["type"] = "svn_copy";
+            item["path"] = path;
+            item["orig_path"] = origin.Path;
+            item["orig_revision"] = origin.Revision;
+            Actions_->AppendValue(item);
+        }
+    }
+
+    void MkDir(const TString& path) {
+        if (Actions_) {
+            TJsonValue item;
+            item["type"] = "mkdir";
+            item["path"] = path;
+            Actions_->AppendValue(item);
+        }
+    }
+
+    void RemoveFile(const TString& path) {
+        if (Actions_) {
+            TJsonValue item;
+            item["type"] = "remove_file";
+            item["path"] = path;
+            Actions_->AppendValue(item);
+        }
+    }
+
+    void RemoveTree(const TString& path) {
+        if (Actions_) {
+            TJsonValue item;
+            item["type"] = "remove_tree";
+            item["path"] = path;
+            Actions_->AppendValue(item);
+        }
+    }
+
+    void StoreFile(
+        const TString& path,
+        const TString& data,
+        const bool execute,
+        const bool symlink,
+        const TMaybe<bool> binaryHint,
+        const TMaybe<bool> encrypted)
+    {
+        if (Actions_) {
+            const TString file = Join("/", "files", path);
+            TJsonValue item;
+            item["type"] = "store_file";
+            item["executable"] = execute;
+            item["path"] = path;
+            item["file"] = file;
+            if (binaryHint.Defined()) {
+                item["binary_hint"] = *binaryHint;
+            }
+            if (encrypted.Defined()) {
+                item["encrypted"] = *encrypted;
+            }
+            Actions_->AppendValue(item);
+            WriteEntry(file, data, symlink);
+        }
+    }
+
+    void SetBaseSvnRevision(ui64 revision) {
+        if (Meta_) {
+            (*Meta_)["base_svn_revision"] = revision;
+        }
+    }
+
+    void AddRevprop(const TString& prop, const TString& value) {
+        if (Revprops_) {
+            (*Revprops_)[prop] = value;
+        }
+    }
+
+private:
+    void WriteEntry(
+        const TString& path,
+        const TString& data,
+        const bool symlink = false)
+    {
+        struct archive_entry* const entry = archive_entry_new();
+        // Write header.
+        archive_entry_set_pathname(entry, path.c_str());
+        archive_entry_set_size(entry, data.size());
+        archive_entry_set_filetype(entry, symlink ? AE_IFLNK : AE_IFREG);
+        archive_entry_set_perm(entry, 0644);
+        if (symlink) {
+            archive_entry_set_symlink(entry, data.c_str());
+        }
+        archive_write_header(Archive_, entry);
+        // Write data.
+        // If entry is symlink then entry size become zero.
+        if (archive_entry_size(entry) > 0) {
+            archive_write_data(Archive_, data.data(), data.size());
+        }
+        archive_entry_free(entry);
+    }
+
+private:
+    THolder<NJson::TJsonValue> Actions_;
+    THolder<NJson::TJsonValue> Meta_;
+    THolder<NJson::TJsonValue> Revprops_;
+    struct archive* Archive_;
+};
+
+TWriter::TWriter(const TFsPath& path)
+    : Impl_(new TImpl(path))
+{
+}
+
+TWriter::~TWriter()
+{ }
+
+void TWriter::Finish() {
+    Impl_->Finish();
+}
+
+void TWriter::SetBaseSvnRevision(ui64 revision) {
+    Impl_->SetBaseSvnRevision(revision);
+}
+
+void TWriter::AddRevprop(const TString& prop, const TString& value) {
+    Impl_->AddRevprop(prop, value);
+}
+
+void TWriter::Copy(const TString& path, const TOrigin& origin) {
+    Impl_->Copy(path, origin);
+}
+
+void TWriter::MkDir(const TString& path) {
+    Impl_->MkDir(path);
+}
+
+void TWriter::RemoveFile(const TString& path) {
+    Impl_->RemoveFile(path);
+}
+
+void TWriter::RemoveTree(const TString& path) {
+    Impl_->RemoveTree(path);
+}
+
+void TWriter::StoreFile(
+    const TString& path,
+    const TString& data,
+    const bool execute,
+    const bool symlink,
+    const TMaybe<bool> binaryHint,
+    const TMaybe<bool> encrypted)
+{
+    Impl_->StoreFile(path, data, execute, symlink, binaryHint, encrypted);
+}
+
+} // namespace NZipatch
+
diff --git a/library/cpp/zipatch/writer.h b/library/cpp/zipatch/writer.h
new file mode 100644
index 0000000000..75cbe49777
--- /dev/null
+++ b/library/cpp/zipatch/writer.h
@@ -0,0 +1,51 @@
+#pragma once
+
+#include <util/folder/path.h>
+#include <util/generic/ptr.h>
+#include <util/generic/maybe.h>
+
+namespace NZipatch {
+
+class TWriter {
+public:
+    struct TOrigin {
+        TString Path;
+        ui64 Revision;
+
+        inline TOrigin(const TString& path, const ui64 revision)
+            : Path(path)
+            , Revision(revision)
+        { }
+    };
+
+     TWriter(const TFsPath& path);
+    ~TWriter();
+
+    void Finish();
+
+    void SetBaseSvnRevision(ui64 revision);
+
+    void AddRevprop(const TString& prop, const TString& value);
+
+    void Copy(const TString& path, const TOrigin& origin);
+
+    void MkDir(const TString& path);
+
+    void RemoveFile(const TString& path);
+
+    void RemoveTree(const TString& path);
+
+    void StoreFile(const TString& path,
+                   const TString& data,
+                   const bool execute,
+                   const bool symlink,
+                   const TMaybe<bool> binaryHint = Nothing(),
+                   const TMaybe<bool> encrypted = Nothing());
+
+private:
+    class TImpl;
+    THolder<TImpl> Impl_;
+};
+
+} // namespace NZipatch
+
diff --git a/library/cpp/zipatch/ya.make b/library/cpp/zipatch/ya.make
new file mode 100644
index 0000000000..f8fd6006b2
--- /dev/null
+++ b/library/cpp/zipatch/ya.make
@@ -0,0 +1,16 @@
+LIBRARY()
+
+SRCS(
+    reader.cpp
+    writer.cpp
+)
+
+PEERDIR(
+    contrib/libs/libarchive
+    library/cpp/json
+)
+
+GENERATE_ENUM_SERIALIZATION(reader.h)
+
+END()
+