<commit_before>package worker_integration_test

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"sort"
	"strings"
	"testing"

	"github.com/streadway/amqp"
)

type LogPart struct {
	ID      int64  `json:"id"`
	Content string `json:"log"`
	Final   bool   `json:"final"`
	UUID    string `json:"uuid"`
	Number  int64  `json:"number"`
}

type LogPartSlice []LogPart

func (lps LogPartSlice) Len() int {
	return len(lps)
}

func (lps LogPartSlice) Less(i, j int) bool {
	return lps[i].Number < lps[j].Number
}

func (lps LogPartSlice) Swap(i, j int) {
	lps[i], lps[j] = lps[j], lps[i]
}

type StateUpdate struct {
	ID    int64  `json:"id"`
	State string `json:"state"`
}

func setupConn() (*amqp.Connection, error) {
	amqpConn, err := amqp.Dial(os.Getenv("AMQP_URL"))
	if err != nil {
		return nil, err
	}

	amqpChan, err := amqpConn.Channel()
	if err != nil {
		return nil, err
	}
	defer amqpChan.Close()

	_, err = amqpChan.QueueDeclare("builds.test", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueueDeclare("reporting.jobs.logs", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueueDeclare("reporting.jobs.builds", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueuePurge("builds.test", false)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueuePurge("reporting.jobs.logs", false)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueuePurge("reporting.jobs.builds", false)
	if err != nil {
		return nil, err
	}

	return amqpConn, nil
}

func publishJob(amqpConn *amqp.Connection) error {
	amqpChan, err := amqpConn.Channel()
	if err != nil {
		return err
	}
	defer amqpChan.Close()

	return amqpChan.Publish("", "builds.test", false, false, amqp.Publishing{
		Body: []byte(`{
			"type": "test",
			"job": {
				"id": 3,
				"number": "1.1",
				"commit": "abcdef",
				"commit_range": "abcde...abcdef",
				"commit_message": "Hello world",
				"branch": "master",
				"ref": null,
				"state": "queued",
				"secure_env_enabled": true,
				"pull_request": false
			},
			"source": {
				"id": 2,
				"number": "1"
			},
			"repository": {
				"id": 1,
				"slug": "hello/world",
				"github_id": 1234,
				"source_url": "git://github.com/hello/world.git",
				"api_url": "https://api.github.com",
				"last_build_id": 2,
				"last_build_number": "1",
				"last_build_started_at": null,
				"last_build_finished_at": null,
				"last_build_duration": null,
				"last_build_state": "created",
				"description": "Hello world",
				"config": {},
				"queue": "builds.test",
				"uuid": "fake-uuid",
				"ssh_key": null,
				"env_vars": [],
				"timeouts": {
					"hard_limit": null,
					"log_silence": null
				}
			}`),
		DeliveryMode: amqp.Persistent,
	})
}

func TestIntegrationLogMessages(t *testing.T) {
	if os.Getenv("AMQP_URI") == "" {
		t.Skip("Skipping integration test as AMQP_URI isn't set")
	}

	buildScriptServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprintln(w, "Hello, world")
	}))
	defer buildScriptServer.Close()

	conn, err := setupConn()
	if err != nil {
		t.Fatal(err)
	}

	err = publishJob(conn)
	if err != nil {
		t.Fatal(err)
	}

	amqpChan, err := conn.Channel()
	if err != nil {
		t.Fatal(err)
	}

	logParts := make([]LogPart, 2)

	delivery, _, err := amqpChan.Get("reporting.jobs.logs", true)
	if err != nil {
		t.Fatal(err)
	}

	err = json.Unmarshal(delivery.Body, logParts[0])
	if err != nil {
		t.Fatal(err)<commit_msg>Conditionally disable integration tests until further work can be done<commit_after>package worker_integration_test

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"sort"
	"strings"
	"testing"

	"github.com/streadway/amqp"
)

type LogPart struct {
	ID      int64  `json:"id"`
	Content string `json:"log"`
	Final   bool   `json:"final"`
	UUID    string `json:"uuid"`
	Number  int64  `json:"number"`
}

type LogPartSlice []LogPart

func (lps LogPartSlice) Len() int {
	return len(lps)
}

func (lps LogPartSlice) Less(i, j int) bool {
	return lps[i].Number < lps[j].Number
}

func (lps LogPartSlice) Swap(i, j int) {
	lps[i], lps[j] = lps[j], lps[i]
}

type StateUpdate struct {
	ID    int64  `json:"id"`
	State string `json:"state"`
}

func setupConn() (*amqp.Connection, error) {
	amqpConn, err := amqp.Dial(os.Getenv("AMQP_URL"))
	if err != nil {
		return nil, err
	}

	amqpChan, err := amqpConn.Channel()
	if err != nil {
		return nil, err
	}
	defer amqpChan.Close()

	_, err = amqpChan.QueueDeclare("builds.test", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueueDeclare("reporting.jobs.logs", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueueDeclare("reporting.jobs.builds", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueuePurge("builds.test", false)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueuePurge("reporting.jobs.logs", false)
	if err != nil {
		return nil, err
	}

	_, err = amqpChan.QueuePurge("reporting.jobs.builds", false)
	if err != nil {
		return nil, err
	}

	return amqpConn, nil
}

func publishJob(amqpConn *amqp.Connection) error {
	amqpChan, err := amqpConn.Channel()
	if err != nil {
		return err
	}
	defer amqpChan.Close()

	return amqpChan.Publish("", "builds.test", false, false, amqp.Publishing{
		Body: []byte(`{
			"type": "test",
			"job": {
				"id": 3,
				"number": "1.1",
				"commit": "abcdef",
				"commit_range": "abcde...abcdef",
				"commit_message": "Hello world",
				"branch": "master",
				"ref": null,
				"state": "queued",
				"secure_env_enabled": true,
				"pull_request": false
			},
			"source": {
				"id": 2,
				"number": "1"
			},
			"repository": {
				"id": 1,
				"slug": "hello/world",
				"github_id": 1234,
				"source_url": "git://github.com/hello/world.git",
				"api_url": "https://api.github.com",
				"last_build_id": 2,
				"last_build_number": "1",
				"last_build_started_at": null,
				"last_build_finished_at": null,
				"last_build_duration": null,
				"last_build_state": "created",
				"description": "Hello world",
				"config": {},
				"queue": "builds.test",
				"uuid": "fake-uuid",
				"ssh_key": null,
				"env_vars": [],
				"timeouts": {
					"hard_limit": null,
					"log_silence": null
				}
			}`),
		DeliveryMode: amqp.Persistent,
	})
}

func TestIntegrationLogMessages(t *testing.T) {
	if os.Getenv("AMQP_URI") == "" {
		t.Skip("Skipping integration test as AMQP_URI isn't set")
	}

	if os.Getenv("INTEGRATION_TESTS_DISABLED") != "" {
		t.Skip("Skipping disabled integration tests")
	}

	buildScriptServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprintln(w, "Hello, world")
	}))
	defer buildScriptServer.Close()

	conn, err := setupConn()
	if err != nil {
		t.Fatal(err)
	}

	err = publishJob(conn)
	if err != nil {
		t.Fatal(err)
	}

	amqpChan, err := conn.Channel()
	if err != nil {
		t.Fatal(err)
	}

	logParts := make([]LogPart, 2)

	delivery, _, err := amqpChan.Get("reporting.jobs.logs", true)
	if err != nil {
		t.Fatal(err)
	}

	err = json.Unmarshal(delivery.Body, logParts[0])
	if err != nil {
		t.Fatal(err)
{'fraction_non_alphanumeric': 0.1723430447271235, 'fraction_numerical': 0.005471207769115032, 'mean_word_length': 3.5219542362399507, 'pattern_counts': {'":': 72, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3600498', 'n_tokens_mistral': 3231, 'n_tokens_neox': 2836, 'n_words': 642}