c/ooni-probe-cli
Watch
1
0
Fork
You've already forked ooni-probe-cli
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

chore: merge probe-engine into probe-cli (#201)

Browse source 
This is how I did it:

1. `git clone https://github.com/ooni/probe-engine internal/engine`

2. ```
(cd internal/engine && git describe --tags)
v0.23.0
```

3. `nvim go.mod` (merging `go.mod` with `internal/engine/go.mod`

4. `rm -rf internal/.git internal/engine/go.{mod,sum}`

5. `git add internal/engine`

6. `find . -type f -name \*.go -exec sed -i 's@/ooni/probe-engine@/ooni/probe-cli/v3/internal/engine@g' {} \;`

7. `go build ./...` (passes)

8. `go test -race ./...` (temporary failure on RiseupVPN)

9. `go mod tidy`

10. this commit message

Once this piece of work is done, we can build a new version of `ooniprobe` that
is using `internal/engine` directly. We need to do more work to ensure all the
other functionality in `probe-engine` (e.g. making mobile packages) are still WAI.

Part of https://github.com/ooni/probe/issues/1335
This commit is contained in:
Simone Basso 2021-02-02 12:05:47 +01:00 • committed by GitHub
commit d57c78bc71
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
535 changed files with 66182 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

20
internal/engine/oonimkall/README.md Normal file
View file

@ -0,0 +1,20 @@
# Package github.com/ooni/probe-engine/oonimkall
Package oonimkall implements APIs used by OONI mobile apps. We
expose these APIs to mobile apps using gomobile.
We expose two APIs: the task API, which is derived from the
API originally exposed by Measurement Kit, and the session API,
which is a Go API that mobile apps can use via `gomobile`.
This package is named oonimkall because it contains a partial
reimplementation of the mkall API implemented by Measurement Kit
in, e.g., [mkall-ios](https://github.com/measurement-kit/mkall-ios).
The basic tenet of the task API is that you define an experiment
task you wanna run using a JSON, then you start a task for it, and
you receive events as serialized JSONs. In addition to this
functionality, we also include extra APIs used by OONI mobile.
The basic tenet of the session API is that you create an instance
of `Session` and use it to perform the operations you need.

377
internal/engine/oonimkall/session.go Normal file
View file

@ -0,0 +1,377 @@
package oonimkall
import (
"context"
"encoding/json"
"errors"
"fmt"
"runtime"
"sync"
engine "github.com/ooni/probe-cli/v3/internal/engine"
"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
"github.com/ooni/probe-cli/v3/internal/engine/internal/runtimex"
"github.com/ooni/probe-cli/v3/internal/engine/model"
"github.com/ooni/probe-cli/v3/internal/engine/probeservices"
)
// AtomicInt64 allows us to export atomicx.Int64 variables to
// mobile libraries so we can use them in testing.
type AtomicInt64 struct {
*atomicx.Int64
}
// The following two variables contain metrics pertaining to the number
// of Sessions and Contexts that are currently being used.
var (
ActiveSessions = &AtomicInt64{atomicx.NewInt64()}
ActiveContexts = &AtomicInt64{atomicx.NewInt64()}
)
// Logger is the logger used by a Session. You should implement a class
// compatible with this interface in Java/ObjC and then save a reference
// to this instance in the SessionConfig object. All log messages that
// the Session will generate will be routed to this Logger.
type Logger interface {
Debug(msg string)
Info(msg string)
Warn(msg string)
}
// SessionConfig contains configuration for a Session. You should
// fill all the mandatory fields and could also optionally fill some of
// the optional fields. Then pass this struct to NewSession.
type SessionConfig struct {
// AssetsDir is the mandatory directory where to store assets
// required by a Session, e.g. MaxMind DB files.
AssetsDir string
// Logger is the optional logger that will receive all the
// log messages generated by a Session. If this field is nil
// then the session will not emit any log message.
Logger Logger
// ProbeServicesURL allows you to optionally force the
// usage of an alternative probe service instance. This setting
// should only be used for implementing integration tests.
ProbeServicesURL string
// SoftwareName is the mandatory name of the application
// that will be using the new Session.
SoftwareName string
// SoftwareVersion is the mandatory version of the application
// that will be using the new Session.
SoftwareVersion string
// StateDir is the mandatory directory where to store state
// information required by a Session.
StateDir string
// TempDir is the mandatory directory where the Session shall
// store temporary files. Among other tasks, Session.Close will
// remove any temporary file created within this Session.
TempDir string
// Verbose is optional. If there is a non-null Logger and this
// field is true, then the Logger will also receive Debug messages,
// otherwise it will not receive such messages.
Verbose bool
}
// Session contains shared state for running experiments and/or other
// OONI related task (e.g. geolocation). Note that the Session isn't
// mean to be a long living object. The workflow is to create a Session,
// do the operations you need to do with it now, then make sure it is
// not referenced by other variables, so the Go GC can finalize it.
//
// Future directions
//
// We will eventually rewrite the code for running new experiments such
// that a Task will be created from a Session, such that experiments
// could share the same Session and save geolookups, etc. For now, we
// are in the suboptimal situations where Tasks create, use, and close
// their own session, thus running more lookups than needed.
type Session struct {
cl []context.CancelFunc
mtx sync.Mutex
submitter *probeservices.Submitter
sessp *engine.Session
// Hooks for testing (should not appear in Java/ObjC)
TestingCheckInBeforeNewProbeServicesClient func(ctx *Context)
TestingCheckInBeforeCheckIn func(ctx *Context)
}
// NewSession creates a new session. You should use a session for running
// a set of operations in a relatively short time frame. You SHOULD NOT create
// a single session and keep it all alive for the whole app lifecyle, since
// the Session code is not specifically designed for this use case.
func NewSession(config *SessionConfig) (*Session, error) {
kvstore, err := engine.NewFileSystemKVStore(config.StateDir)
if err != nil {
return nil, err
}
var availableps []model.Service
if config.ProbeServicesURL != "" {
availableps = append(availableps, model.Service{
Address: config.ProbeServicesURL,
Type: "https",
})
}
engineConfig := engine.SessionConfig{
AssetsDir: config.AssetsDir,
AvailableProbeServices: availableps,
KVStore: kvstore,
Logger: newLogger(config.Logger, config.Verbose),
SoftwareName: config.SoftwareName,
SoftwareVersion: config.SoftwareVersion,
TempDir: config.TempDir,
}
sessp, err := engine.NewSession(engineConfig)
if err != nil {
return nil, err
}
sess := &Session{sessp: sessp}
runtime.SetFinalizer(sess, sessionFinalizer)
ActiveSessions.Add(1)
return sess, nil
}
// sessionFinalizer finalizes a Session. While in general in Go code using a
// finalizer is probably unclean, it seems that using a finalizer when binding
// with Java/ObjC code is actually useful to simplify the apps.
func sessionFinalizer(sess *Session) {
for _, fn := range sess.cl {
fn()
}
sess.sessp.Close() // ignore return value
ActiveSessions.Add(-1)
}
// Context is the context of an operation. You use this context
// to cancel a long running operation by calling Cancel(). Because
// you create a Context from a Session and because the Session is
// keeping track of the Context instances it owns, you do don't
// need to call the Cancel method when you're done.
type Context struct {
cancel context.CancelFunc
ctx context.Context
}
// Cancel cancels pending operations using this context.
func (ctx *Context) Cancel() {
ctx.cancel()
}
// NewContext creates an new interruptible Context.
func (sess *Session) NewContext() *Context {
return sess.NewContextWithTimeout(-1)
}
// NewContextWithTimeout creates an new interruptible Context that will automatically
// cancel itself after the given timeout. Setting a zero or negative timeout implies
// there is no actual timeout configured for the Context.
func (sess *Session) NewContextWithTimeout(timeout int64) *Context {
sess.mtx.Lock()
defer sess.mtx.Unlock()
ctx, origcancel := newContext(timeout)
ActiveContexts.Add(1)
var once sync.Once
cancel := func() {
once.Do(func() {
ActiveContexts.Add(-1)
origcancel()
})
}
sess.cl = append(sess.cl, cancel)
return &Context{cancel: cancel, ctx: ctx}
}
// GeolocateResults contains the GeolocateTask results.
type GeolocateResults struct {
// ASN is the autonomous system number.
ASN string
// Country is the country code.
Country string
// IP is the IP address.
IP string
// Org is the commercial name of the ASN.
Org string
}
// MaybeUpdateResources ensures that resources are up to date.
func (sess *Session) MaybeUpdateResources(ctx *Context) error {
sess.mtx.Lock()
defer sess.mtx.Unlock()
return sess.sessp.MaybeUpdateResources(ctx.ctx)
}
// Geolocate performs a geolocate operation and returns the results. This method
// is (in Java terminology) synchronized with the session instance.
func (sess *Session) Geolocate(ctx *Context) (*GeolocateResults, error) {
sess.mtx.Lock()
defer sess.mtx.Unlock()
info, err := sess.sessp.LookupLocationContext(ctx.ctx)
if err != nil {
return nil, err
}
return &GeolocateResults{
ASN: fmt.Sprintf("AS%d", info.ASN),
Country: info.CountryCode,
IP: info.ProbeIP,
Org: info.NetworkName,
}, nil
}
// SubmitMeasurementResults contains the results of a single measurement submission
// to the OONI backends using the OONI collector API.
type SubmitMeasurementResults struct {
UpdatedMeasurement string
UpdatedReportID string
}
// Submit submits the given measurement and returns the results. This method is (in
// Java terminology) synchronized with the Session instance.
func (sess *Session) Submit(ctx *Context, measurement string) (*SubmitMeasurementResults, error) {
sess.mtx.Lock()
defer sess.mtx.Unlock()
if sess.submitter == nil {
psc, err := sess.sessp.NewProbeServicesClient(ctx.ctx)
if err != nil {
return nil, err
}
sess.submitter = probeservices.NewSubmitter(psc, sess.sessp.Logger())
}
var mm model.Measurement
if err := json.Unmarshal([]byte(measurement), &mm); err != nil {
return nil, err
}
if err := sess.submitter.Submit(ctx.ctx, &mm); err != nil {
return nil, err
}
data, err := json.Marshal(mm)
runtimex.PanicOnError(err, "json.Marshal should not fail here")
return &SubmitMeasurementResults{
UpdatedMeasurement: string(data),
UpdatedReportID: mm.ReportID,
}, nil
}
// CheckInConfigWebConnectivity is the configuration for the WebConnectivity test
type CheckInConfigWebConnectivity struct {
CategoryCodes []string // CategoryCodes is an array of category codes
}
// Add a category code to the array in CheckInConfigWebConnectivity
func (ckw *CheckInConfigWebConnectivity) Add(cat string) {
ckw.CategoryCodes = append(ckw.CategoryCodes, cat)
}
func (ckw *CheckInConfigWebConnectivity) toModel() model.CheckInConfigWebConnectivity {
return model.CheckInConfigWebConnectivity{
CategoryCodes: ckw.CategoryCodes,
}
}
// CheckInConfig contains configuration for calling the checkin API.
type CheckInConfig struct {
Charging bool // Charging indicate if the phone is actually charging
OnWiFi bool // OnWiFi indicate if the phone is actually connected to a WiFi network
Platform string // Platform of the probe
RunType string // RunType
SoftwareName string // SoftwareName of the probe
SoftwareVersion string // SoftwareVersion of the probe
WebConnectivity *CheckInConfigWebConnectivity // WebConnectivity class contain an array of categories
}
// CheckInInfoWebConnectivity contains the array of URLs returned by the checkin API
type CheckInInfoWebConnectivity struct {
ReportID string
URLs []model.URLInfo
}
// URLInfo contains info on a test lists URL
type URLInfo struct {
CategoryCode string
CountryCode string
URL string
}
// Size returns the number of URLs.
func (ckw *CheckInInfoWebConnectivity) Size() int64 {
return int64(len(ckw.URLs))
}
// At gets the URLInfo at position idx from CheckInInfoWebConnectivity.URLs
func (ckw *CheckInInfoWebConnectivity) At(idx int64) *URLInfo {
if idx < 0 || int(idx) >= len(ckw.URLs) {
return nil
}
w := ckw.URLs[idx]
return &URLInfo{
CategoryCode: w.CategoryCode,
CountryCode: w.CountryCode,
URL: w.URL,
}
}
func newCheckInInfoWebConnectivity(ckw *model.CheckInInfoWebConnectivity) *CheckInInfoWebConnectivity {
if ckw == nil {
return nil
}
out := new(CheckInInfoWebConnectivity)
out.ReportID = ckw.ReportID
out.URLs = ckw.URLs
return out
}
// CheckInInfo contains the return test objects from the checkin API
type CheckInInfo struct {
WebConnectivity *CheckInInfoWebConnectivity
}
// CheckIn function is called by probes asking if there are tests to be run
// The config argument contains the mandatory settings.
// Returns the list of tests to run and the URLs, on success, or an explanatory error, in case of failure.
func (sess *Session) CheckIn(ctx *Context, config *CheckInConfig) (*CheckInInfo, error) {
sess.mtx.Lock()
defer sess.mtx.Unlock()
if config.WebConnectivity == nil {
return nil, errors.New("oonimkall: missing webconnectivity config")
}
info, err := sess.sessp.LookupLocationContext(ctx.ctx)
if err != nil {
return nil, err
}
if sess.TestingCheckInBeforeNewProbeServicesClient != nil {
sess.TestingCheckInBeforeNewProbeServicesClient(ctx)
}
psc, err := sess.sessp.NewProbeServicesClient(ctx.ctx)
if err != nil {
return nil, err
}
if sess.TestingCheckInBeforeCheckIn != nil {
sess.TestingCheckInBeforeCheckIn(ctx)
}
cfg := model.CheckInConfig{
Charging: config.Charging,
OnWiFi: config.OnWiFi,
Platform: config.Platform,
ProbeASN: info.ASNString(),
ProbeCC: info.CountryCode,
RunType: config.RunType,
SoftwareVersion: config.SoftwareVersion,
WebConnectivity: config.WebConnectivity.toModel(),
}
result, err := psc.CheckIn(ctx.ctx, cfg)
if err != nil {
return nil, err
}
return &CheckInInfo{
WebConnectivity: newCheckInInfoWebConnectivity(result.WebConnectivity),
}, nil
}

440
internal/engine/oonimkall/session_integration_test.go Normal file
View file

@ -0,0 +1,440 @@
package oonimkall_test
import (
"context"
"encoding/json"
"errors"
"fmt"
"io/ioutil"
"os"
"runtime"
"strings"
"testing"
"time"
engine "github.com/ooni/probe-cli/v3/internal/engine"
"github.com/ooni/probe-cli/v3/internal/engine/geolocate"
"github.com/ooni/probe-cli/v3/internal/engine/model"
"github.com/ooni/probe-cli/v3/internal/engine/oonimkall"
)
func NewSessionWithAssetsDir(assetsDir string) (*oonimkall.Session, error) {
return oonimkall.NewSession(&oonimkall.SessionConfig{
AssetsDir: assetsDir,
ProbeServicesURL: "https://ams-pg-test.ooni.org/",
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
StateDir: "../testdata/oonimkall/state",
TempDir: "../testdata/",
})
}
func NewSession() (*oonimkall.Session, error) {
return NewSessionWithAssetsDir("../testdata/oonimkall/assets")
}
func TestNewSessionWithInvalidStateDir(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
sess, err := oonimkall.NewSession(&oonimkall.SessionConfig{
StateDir: "",
})
if err == nil || !strings.HasSuffix(err.Error(), "no such file or directory") {
t.Fatal("not the error we expected")
}
if sess != nil {
t.Fatal("expected a nil Session here")
}
}
func TestNewSessionWithMissingSoftwareName(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
sess, err := oonimkall.NewSession(&oonimkall.SessionConfig{
StateDir: "../testdata/oonimkall/state",
})
if err == nil || err.Error() != "AssetsDir is empty" {
t.Fatal("not the error we expected")
}
if sess != nil {
t.Fatal("expected a nil Session here")
}
}
func TestMaybeUpdateResourcesWithCancelledContext(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
dir, err := ioutil.TempDir("", "xx")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(dir)
sess, err := NewSessionWithAssetsDir(dir)
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
ctx.Cancel() // cause immediate failure
err = sess.MaybeUpdateResources(ctx)
if !errors.Is(err, context.Canceled) {
t.Fatalf("not the error we expected: %+v", err)
}
}
func ReduceErrorForGeolocate(err error) error {
if err == nil {
return errors.New("we expected an error here")
}
if errors.Is(err, context.Canceled) {
return nil // when we have not downloaded the resources yet
}
if !errors.Is(err, geolocate.ErrAllIPLookuppersFailed) {
return nil // otherwise
}
return fmt.Errorf("not the error we expected: %w", err)
}
func TestGeolocateWithCancelledContext(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
ctx.Cancel() // cause immediate failure
location, err := sess.Geolocate(ctx)
if err := ReduceErrorForGeolocate(err); err != nil {
t.Fatal(err)
}
if location != nil {
t.Fatal("expected nil location here")
}
}
func TestGeolocateGood(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
location, err := sess.Geolocate(ctx)
if err != nil {
t.Fatal(err)
}
if location.ASN == "" {
t.Fatal("location.ASN is empty")
}
if location.Country == "" {
t.Fatal("location.Country is empty")
}
if location.IP == "" {
t.Fatal("location.IP is empty")
}
if location.Org == "" {
t.Fatal("location.Org is empty")
}
}
func ReduceErrorForSubmitter(err error) error {
if err == nil {
return errors.New("we expected an error here")
}
if errors.Is(err, context.Canceled) {
return nil // when we have not downloaded the resources yet
}
if err.Error() == "all available probe services failed" {
return nil // otherwise
}
return fmt.Errorf("not the error we expected: %w", err)
}
func TestSubmitWithCancelledContext(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
ctx.Cancel() // cause immediate failure
result, err := sess.Submit(ctx, "{}")
if err := ReduceErrorForSubmitter(err); err != nil {
t.Fatalf("not the error we expected: %+v", err)
}
if result != nil {
t.Fatal("expected nil result here")
}
}
func TestSubmitWithInvalidJSON(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
result, err := sess.Submit(ctx, "{")
if err == nil || err.Error() != "unexpected end of JSON input" {
t.Fatalf("not the error we expected: %+v", err)
}
if result != nil {
t.Fatal("expected nil result here")
}
}
func DoSubmission(ctx *oonimkall.Context, sess *oonimkall.Session) error {
inputm := model.Measurement{
DataFormatVersion: "0.2.0",
MeasurementStartTime: "2019-10-28 12:51:07",
MeasurementRuntime: 1.71,
ProbeASN: "AS30722",
ProbeCC: "IT",
ProbeIP: "127.0.0.1",
ReportID: "",
ResolverIP: "172.217.33.129",
SoftwareName: "miniooni",
SoftwareVersion: "0.1.0-dev",
TestKeys: map[string]bool{"success": true},
TestName: "example",
TestVersion: "0.1.0",
}
inputd, err := json.Marshal(inputm)
if err != nil {
return err
}
result, err := sess.Submit(ctx, string(inputd))
if err != nil {
return fmt.Errorf("session_test.go: submit failed: %w", err)
}
if result.UpdatedMeasurement == "" {
return errors.New("expected non empty measurement")
}
if result.UpdatedReportID == "" {
return errors.New("expected non empty report ID")
}
var outputm model.Measurement
return json.Unmarshal([]byte(result.UpdatedMeasurement), &outputm)
}
func TestSubmitMeasurementGood(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
if err := DoSubmission(ctx, sess); err != nil {
t.Fatal(err)
}
}
func TestSubmitCancelContextAfterFirstSubmission(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
if err := DoSubmission(ctx, sess); err != nil {
t.Fatal(err)
}
ctx.Cancel() // fail second submission
err = DoSubmission(ctx, sess)
if err == nil || !strings.HasPrefix(err.Error(), "session_test.go: submit failed") {
t.Fatalf("not the error we expected: %+v", err)
}
if !errors.Is(err, context.Canceled) {
t.Fatalf("not the error we expected: %+v", err)
}
}
func TestCheckInSuccess(t *testing.T) {
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
config := oonimkall.CheckInConfig{
Charging: true,
OnWiFi: true,
Platform: "android",
RunType: "timed",
SoftwareName: "ooniprobe-android",
SoftwareVersion: "2.7.1",
WebConnectivity: &oonimkall.CheckInConfigWebConnectivity{},
}
config.WebConnectivity.Add("NEWS")
config.WebConnectivity.Add("CULTR")
result, err := sess.CheckIn(ctx, &config)
if err != nil {
t.Fatalf("unexpected error: %+v", err)
}
if result == nil || result.WebConnectivity == nil {
t.Fatal("got nil result or WebConnectivity")
}
if len(result.WebConnectivity.URLs) < 1 {
t.Fatal("unexpected number of URLs")
}
if result.WebConnectivity.ReportID == "" {
t.Fatal("got empty report ID")
}
siz := result.WebConnectivity.Size()
if siz <= 0 {
t.Fatal("unexpected number of URLs")
}
for idx := int64(0); idx < siz; idx++ {
entry := result.WebConnectivity.At(idx)
if entry.CategoryCode != "NEWS" && entry.CategoryCode != "CULTR" {
t.Fatalf("unexpected category code: %+v", entry)
}
}
if result.WebConnectivity.At(-1) != nil {
t.Fatal("expected nil here")
}
if result.WebConnectivity.At(siz) != nil {
t.Fatal("expected nil here")
}
}
func TestCheckInLookupLocationFailure(t *testing.T) {
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
config := oonimkall.CheckInConfig{
Charging: true,
OnWiFi: true,
Platform: "android",
RunType: "timed",
SoftwareName: "ooniprobe-android",
SoftwareVersion: "2.7.1",
WebConnectivity: &oonimkall.CheckInConfigWebConnectivity{},
}
config.WebConnectivity.Add("NEWS")
config.WebConnectivity.Add("CULTR")
ctx.Cancel() // immediate failure
result, err := sess.CheckIn(ctx, &config)
if !errors.Is(err, geolocate.ErrAllIPLookuppersFailed) {
t.Fatalf("not the error we expected: %+v", err)
}
if result != nil {
t.Fatal("expected nil result here")
}
}
func TestCheckInNewProbeServicesFailure(t *testing.T) {
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
sess.TestingCheckInBeforeNewProbeServicesClient = func(ctx *oonimkall.Context) {
ctx.Cancel() // cancel execution
}
ctx := sess.NewContext()
config := oonimkall.CheckInConfig{
Charging: true,
OnWiFi: true,
Platform: "android",
RunType: "timed",
SoftwareName: "ooniprobe-android",
SoftwareVersion: "2.7.1",
WebConnectivity: &oonimkall.CheckInConfigWebConnectivity{},
}
config.WebConnectivity.Add("NEWS")
config.WebConnectivity.Add("CULTR")
result, err := sess.CheckIn(ctx, &config)
if !errors.Is(err, engine.ErrAllProbeServicesFailed) {
t.Fatalf("not the error we expected: %+v", err)
}
if result != nil {
t.Fatal("expected nil result here")
}
}
func TestCheckInCheckInFailure(t *testing.T) {
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
sess.TestingCheckInBeforeCheckIn = func(ctx *oonimkall.Context) {
ctx.Cancel() // cancel execution
}
ctx := sess.NewContext()
config := oonimkall.CheckInConfig{
Charging: true,
OnWiFi: true,
Platform: "android",
RunType: "timed",
SoftwareName: "ooniprobe-android",
SoftwareVersion: "2.7.1",
WebConnectivity: &oonimkall.CheckInConfigWebConnectivity{},
}
config.WebConnectivity.Add("NEWS")
config.WebConnectivity.Add("CULTR")
result, err := sess.CheckIn(ctx, &config)
if !errors.Is(err, context.Canceled) {
t.Fatalf("not the error we expected: %+v", err)
}
if result != nil {
t.Fatal("expected nil result here")
}
}
func TestCheckInNoParams(t *testing.T) {
sess, err := NewSession()
if err != nil {
t.Fatal(err)
}
ctx := sess.NewContext()
config := oonimkall.CheckInConfig{
Charging: true,
OnWiFi: true,
Platform: "android",
RunType: "timed",
SoftwareName: "ooniprobe-android",
SoftwareVersion: "2.7.1",
}
result, err := sess.CheckIn(ctx, &config)
if err == nil || err.Error() != "oonimkall: missing webconnectivity config" {
t.Fatalf("not the error we expected: %+v", err)
}
if result != nil {
t.Fatal("unexpected not nil result here")
}
}
func TestMain(m *testing.M) {
// Here we're basically testing whether eventually the finalizers
// will run and the number of active sessions and cancels will become
// balanced. Especially for the number of active cancels, this is an
// indication that we've correctly cleaned them up in the session.
if exitcode := m.Run(); exitcode != 0 {
os.Exit(exitcode)
}
for {
runtime.GC()
m, n := oonimkall.ActiveContexts.Load(), oonimkall.ActiveSessions.Load()
fmt.Printf("./oonimkall: ActiveContexts: %d; ActiveSessions: %d\n", m, n)
if m == 0 && n == 0 {
break
}
time.Sleep(1 * time.Second)
}
os.Exit(0)
}

10
internal/engine/oonimkall/session_test.go Normal file
View file

@ -0,0 +1,10 @@
package oonimkall
import "testing"
func TestNewCheckInInfoWebConnectivityNilPointer(t *testing.T) {
out := newCheckInInfoWebConnectivity(nil)
if out != nil {
t.Fatal("expected nil pointer")
}
}

29
internal/engine/oonimkall/sessioncontext.go Normal file
View file

@ -0,0 +1,29 @@
package oonimkall
import (
"context"
"math"
"time"
)
const maxTimeout = int64(time.Duration(math.MaxInt64) / time.Second)
func clampTimeout(timeout, max int64) int64 {
if timeout > max {
timeout = max
}
return timeout
}
func newContext(timeout int64) (context.Context, context.CancelFunc) {
return newContextEx(timeout, maxTimeout)
}
func newContextEx(timeout, max int64) (context.Context, context.CancelFunc) {
if timeout > 0 {
timeout = clampTimeout(timeout, max)
return context.WithTimeout(
context.Background(), time.Duration(timeout)*time.Second)
}
return context.WithCancel(context.Background())
}

102
internal/engine/oonimkall/sessioncontext_test.go Normal file
View file

@ -0,0 +1,102 @@
package oonimkall
import (
"testing"
"time"
"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
)
func TestClampTimeout(t *testing.T) {
if clampTimeout(-1, maxTimeout) != -1 {
t.Fatal("unexpected result here")
}
if clampTimeout(0, maxTimeout) != 0 {
t.Fatal("unexpected result here")
}
if clampTimeout(60, maxTimeout) != 60 {
t.Fatal("unexpected result here")
}
if clampTimeout(maxTimeout, maxTimeout) != maxTimeout {
t.Fatal("unexpected result here")
}
if clampTimeout(maxTimeout+1, maxTimeout) != maxTimeout {
t.Fatal("unexpected result here")
}
}
func TestNewContextWithZeroTimeout(t *testing.T) {
here := atomicx.NewInt64()
ctx, cancel := newContext(0)
defer cancel()
go func() {
<-time.After(250 * time.Millisecond)
here.Add(1)
cancel()
}()
<-ctx.Done()
if here.Load() != 1 {
t.Fatal("context timeout not working as intended")
}
}
func TestNewContextWithNegativeTimeout(t *testing.T) {
here := atomicx.NewInt64()
ctx, cancel := newContext(-1)
defer cancel()
go func() {
<-time.After(250 * time.Millisecond)
here.Add(1)
cancel()
}()
<-ctx.Done()
if here.Load() != 1 {
t.Fatal("context timeout not working as intended")
}
}
func TestNewContextWithHugeTimeout(t *testing.T) {
here := atomicx.NewInt64()
ctx, cancel := newContext(maxTimeout + 1)
defer cancel()
go func() {
<-time.After(250 * time.Millisecond)
here.Add(1)
cancel()
}()
<-ctx.Done()
if here.Load() != 1 {
t.Fatal("context timeout not working as intended")
}
}
func TestNewContextWithReasonableTimeout(t *testing.T) {
here := atomicx.NewInt64()
ctx, cancel := newContext(1)
defer cancel()
go func() {
<-time.After(5 * time.Second)
here.Add(1)
cancel()
}()
<-ctx.Done()
if here.Load() != 0 {
t.Fatal("context timeout not working as intended")
}
}
func TestNewContextWithArtificiallyLowMaxTimeout(t *testing.T) {
here := atomicx.NewInt64()
const maxTimeout = 2
ctx, cancel := newContextEx(maxTimeout+1, maxTimeout)
defer cancel()
go func() {
<-time.After(30 * time.Second)
here.Add(1)
cancel()
}()
<-ctx.Done()
if here.Load() != 0 {
t.Fatal("context timeout not working as intended")
}
}

69
internal/engine/oonimkall/sessionlogger.go Normal file
View file

@ -0,0 +1,69 @@
package oonimkall
import (
"fmt"
"github.com/ooni/probe-cli/v3/internal/engine/model"
)
type loggerVerbose struct {
Logger
}
func (slv loggerVerbose) Debugf(format string, v ...interface{}) {
slv.Debug(fmt.Sprintf(format, v...))
}
func (slv loggerVerbose) Infof(format string, v ...interface{}) {
slv.Info(fmt.Sprintf(format, v...))
}
func (slv loggerVerbose) Warnf(format string, v ...interface{}) {
slv.Warn(fmt.Sprintf(format, v...))
}
type loggerNormal struct {
Logger
}
func (sln loggerNormal) Debugf(format string, v ...interface{}) {
// nothing
}
func (sln loggerNormal) Debug(msg string) {
// nothing
}
func (sln loggerNormal) Infof(format string, v ...interface{}) {
sln.Info(fmt.Sprintf(format, v...))
}
func (sln loggerNormal) Warnf(format string, v ...interface{}) {
sln.Warn(fmt.Sprintf(format, v...))
}
type loggerQuiet struct{}
func (loggerQuiet) Debugf(format string, v ...interface{}) {
// nothing
}
func (loggerQuiet) Debug(msg string) {
// nothing
}
func (loggerQuiet) Infof(format string, v ...interface{}) {
// nothing
}
func (loggerQuiet) Info(msg string) {
// nothing
}
func (loggerQuiet) Warnf(format string, v ...interface{}) {
// nothing
}
func (loggerQuiet) Warn(msg string) {
// nothing
}
func newLogger(logger Logger, verbose bool) model.Logger {
if logger == nil {
return loggerQuiet{}
}
if verbose {
return loggerVerbose{Logger: logger}
}
return loggerNormal{Logger: logger}
}

118
internal/engine/oonimkall/sessionlogger_test.go Normal file
View file

@ -0,0 +1,118 @@
package oonimkall
import (
"errors"
"fmt"
"io"
"sync"
"testing"
"github.com/google/go-cmp/cmp"
"github.com/ooni/probe-cli/v3/internal/engine/model"
)
type RecordingLogger struct {
DebugLogs []string
InfoLogs []string
WarnLogs []string
mu sync.Mutex
}
func (rl *RecordingLogger) Debug(msg string) {
rl.mu.Lock()
defer rl.mu.Unlock()
rl.DebugLogs = append(rl.DebugLogs, msg)
}
func (rl *RecordingLogger) Info(msg string) {
rl.mu.Lock()
defer rl.mu.Unlock()
rl.InfoLogs = append(rl.InfoLogs, msg)
}
func (rl *RecordingLogger) Warn(msg string) {
rl.mu.Lock()
defer rl.mu.Unlock()
rl.WarnLogs = append(rl.WarnLogs, msg)
}
func LoggerEmitMessages(logger model.Logger) {
logger.Warnf("a formatted warn message: %+v", io.EOF)
logger.Warn("a warn string")
logger.Infof("a formatted info message: %+v", io.EOF)
logger.Info("a info string")
logger.Debugf("a formatted debug message: %+v", io.EOF)
logger.Debug("a debug string")
}
func TestNewLoggerNilLogger(t *testing.T) {
// The objective of this test is to make sure that, even if the
// Logger instance is nil, we get back something that works, that
// is, something that does not crash when it is used.
logger := newLogger(nil, true)
LoggerEmitMessages(logger)
}
func (rl *RecordingLogger) VerifyNumberOfEntries(debugEntries int) error {
if len(rl.DebugLogs) != debugEntries {
return errors.New("unexpected number of debug messages")
}
if len(rl.InfoLogs) != 2 {
return errors.New("unexpected number of info messages")
}
if len(rl.WarnLogs) != 2 {
return errors.New("unexpected number of warn messages")
}
return nil
}
func (rl *RecordingLogger) ExpectedEntries(level string) []string {
return []string{
fmt.Sprintf("a formatted %s message: %+v", level, io.EOF),
fmt.Sprintf("a %s string", level),
}
}
func (rl *RecordingLogger) CheckNonVerboseEntries() error {
if diff := cmp.Diff(rl.InfoLogs, rl.ExpectedEntries("info")); diff != "" {
return errors.New(diff)
}
if diff := cmp.Diff(rl.WarnLogs, rl.ExpectedEntries("warn")); diff != "" {
return errors.New(diff)
}
return nil
}
func (rl *RecordingLogger) CheckVerboseEntries() error {
if diff := cmp.Diff(rl.DebugLogs, rl.ExpectedEntries("debug")); diff != "" {
return errors.New(diff)
}
return nil
}
func TestNewLoggerQuietLogger(t *testing.T) {
handler := new(RecordingLogger)
logger := newLogger(handler, false)
LoggerEmitMessages(logger)
if err := handler.VerifyNumberOfEntries(0); err != nil {
t.Fatal(err)
}
if err := handler.CheckNonVerboseEntries(); err != nil {
t.Fatal(err)
}
}
func TestNewLoggerVerboseLogger(t *testing.T) {
handler := new(RecordingLogger)
logger := newLogger(handler, true)
LoggerEmitMessages(logger)
if err := handler.VerifyNumberOfEntries(2); err != nil {
t.Fatal(err)
}
if err := handler.CheckNonVerboseEntries(); err != nil {
t.Fatal(err)
}
if err := handler.CheckVerboseEntries(); err != nil {
t.Fatal(err)
}
}

114
internal/engine/oonimkall/task.go Normal file
View file

@ -0,0 +1,114 @@
// Package oonimkall implements APIs used by OONI mobile apps. We
// expose these APIs to mobile apps using gomobile.
//
// We expose two APIs: the task API, which is derived from the
// API originally exposed by Measurement Kit, and the session API,
// which is a Go API that mobile apps can use via `gomobile`.
//
// This package is named oonimkall because it contains a partial
// reimplementation of the mkall API implemented by Measurement Kit
// in, e.g., https://github.com/measurement-kit/mkall-ios.
//
// Task API
//
// The basic tenet of the task API is that you define an experiment
// task you wanna run using a JSON, then you start a task for it, and
// you receive events as serialized JSONs. In addition to this
// functionality, we also include extra APIs used by OONI mobile.
//
// The task API was first defined in Measurement Kit v0.9.0. In this
// context, it was called "the FFI API". The API we expose here is not
// strictly an FFI API, but is close enough for the purpose of using
// OONI from Android and iOS. See https://git.io/Jv4Rv
// (measurement-kit/measurement-kit@v0.10.9) for a comprehensive
// description of MK's FFI API.
//
// See also https://github.com/ooni/probe-cli/v3/internal/engine/pull/347 for the
// design document describing the task API.
//
// See also https://github.com/ooni/probe-cli/v3/internal/engine/blob/master/DESIGN.md,
// which explains why we implemented the oonimkall API.
//
// Session API
//
// The Session API is a Go API that can be exported to mobile apps
// using the gomobile tool. The latest design document for this API is
// at https://github.com/ooni/probe-cli/v3/internal/engine/pull/954.
//
// The basic tenet of the session API is that you create an instance
// of `Session` and use it to perform the operations you need.
package oonimkall
import (
"context"
"encoding/json"
"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
"github.com/ooni/probe-cli/v3/internal/engine/internal/runtimex"
"github.com/ooni/probe-cli/v3/internal/engine/oonimkall/tasks"
)
// Task is an asynchronous task running an experiment. It mimics the
// namesake concept initially implemented in Measurement Kit.
//
// Future directions
//
// Currently Task and Session are two unrelated APIs. As part of
// evolving the APIs with which apps interact with the engine, we
// will modify Task to run in the context of a Session. We will
// do that to save extra lookups and to allow several experiments
// running as subsequent Tasks to reuse the Session connections
// created with the OONI probe services backends.
type Task struct {
cancel context.CancelFunc
isdone *atomicx.Int64
isstopped *atomicx.Int64
out chan *tasks.Event
}
// StartTask starts an asynchronous task. The input argument is a
// serialized JSON conforming to MK v0.10.9's API.
func StartTask(input string) (*Task, error) {
var settings tasks.Settings
if err := json.Unmarshal([]byte(input), &settings); err != nil {
return nil, err
}
const bufsiz = 128 // common case: we don't want runner to block
ctx, cancel := context.WithCancel(context.Background())
task := &Task{
cancel: cancel,
isdone: atomicx.NewInt64(),
isstopped: atomicx.NewInt64(),
out: make(chan *tasks.Event, bufsiz),
}
go func() {
defer close(task.out)
defer task.isstopped.Add(1)
tasks.Run(ctx, &settings, task.out)
}()
return task, nil
}
// WaitForNextEvent blocks until the next event occurs. The returned
// string is a serialized JSON following MK v0.10.9's API.
func (t *Task) WaitForNextEvent() string {
const terminated = `{"key":"task_terminated","value":{}}` // like MK
evp := <-t.out
if evp == nil {
t.isdone.Add(1)
return terminated
}
data, err := json.Marshal(evp)
runtimex.PanicOnError(err, "json.Marshal failed")
return string(data)
}
// IsDone returns true if the task is done.
func (t *Task) IsDone() bool {
return t.isdone.Load() != 0
}
// Interrupt interrupts the task.
func (t *Task) Interrupt() {
t.cancel()
}

560
internal/engine/oonimkall/task_integration_test.go Normal file
View file

@ -0,0 +1,560 @@
package oonimkall_test
import (
"encoding/json"
"errors"
"strings"
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/ooni/probe-cli/v3/internal/engine/model"
"github.com/ooni/probe-cli/v3/internal/engine/oonimkall"
"github.com/ooni/probe-cli/v3/internal/engine/oonimkall/tasks"
)
type eventlike struct {
Key string `json:"key"`
Value map[string]interface{} `json:"value"`
}
func TestGood(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"log_level": "DEBUG",
"name": "Example",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
// interrupt the task so we also exercise this functionality
go func() {
<-time.After(time.Second)
task.Interrupt()
}()
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key == "failure.startup" {
t.Fatal("unexpected failure.startup event")
}
}
// make sure we only see task_terminated at this point
for {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key != "task_terminated" {
t.Fatalf("unexpected event.Key: %s", event.Key)
}
break
}
}
func TestWithMeasurementFailure(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"log_level": "DEBUG",
"name": "ExampleWithFailure",
"options": {
"no_geoip": true,
"no_resolver_lookup": true,
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key == "failure.startup" {
t.Fatal("unexpected failure.startup event")
}
}
}
func TestInvalidJSON(t *testing.T) {
task, err := oonimkall.StartTask(`{`)
var syntaxerr *json.SyntaxError
if !errors.As(err, &syntaxerr) {
t.Fatal("not the expected error")
}
if task != nil {
t.Fatal("task is not nil")
}
}
func TestUnsupportedSetting(t *testing.T) {
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"log_level": "DEBUG",
"name": "Example",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state"
}`)
if err != nil {
t.Fatal(err)
}
var seen bool
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key == "failure.startup" {
if strings.Contains(eventstr, tasks.FailureInvalidVersion) {
seen = true
}
}
}
if !seen {
t.Fatal("did not see failure.startup with invalid version info")
}
}
func TestEmptyStateDir(t *testing.T) {
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"log_level": "DEBUG",
"name": "Example",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
var seen bool
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key == "failure.startup" {
if strings.Contains(eventstr, "mkdir : no such file or directory") {
seen = true
}
}
}
if !seen {
t.Fatal("did not see failure.startup with info that state dir is empty")
}
}
func TestEmptyAssetsDir(t *testing.T) {
task, err := oonimkall.StartTask(`{
"log_level": "DEBUG",
"name": "Example",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
var seen bool
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key == "failure.startup" {
if strings.Contains(eventstr, "AssetsDir is empty") {
seen = true
}
}
}
if !seen {
t.Fatal("did not see failure.startup")
}
}
func TestUnknownExperiment(t *testing.T) {
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"log_level": "DEBUG",
"name": "Antani",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
var seen bool
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key == "failure.startup" {
if strings.Contains(eventstr, "no such experiment: ") {
seen = true
}
}
}
if !seen {
t.Fatal("did not see failure.startup")
}
}
func TestInputIsRequired(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"log_level": "DEBUG",
"name": "ExampleWithInput",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
var seen bool
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key == "failure.startup" {
if strings.Contains(eventstr, "no input provided") {
seen = true
}
}
}
if !seen {
t.Fatal("did not see failure.startup")
}
}
func TestMaxRuntime(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
begin := time.Now()
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"inputs": ["a", "b", "c"],
"name": "ExampleWithInput",
"options": {
"max_runtime": 1,
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key == "failure.startup" {
t.Fatal(eventstr)
}
}
// The runtime is long because of ancillary operations and is even more
// longer because of self shaping we may be performing (especially in
// CI builds) using `-tags shaping`). We have experimentally determined
// that ~10 seconds is the typical CI test run time. See:
//
// 1. https://github.com/ooni/probe-cli/v3/internal/engine/pull/588/checks?check_run_id=667263788
//
// 2. https://github.com/ooni/probe-cli/v3/internal/engine/pull/588/checks?check_run_id=667263855
//
// In case there are further timeouts, e.g. in the sessionresolver, the
// time used by the experiment will be much more. This is for example the
// case in https://github.com/ooni/probe-cli/v3/internal/engine/issues/1005.
if time.Now().Sub(begin) > 10*time.Second {
t.Fatal("expected shorter runtime")
}
}
func TestInterruptExampleWithInput(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
t.Skip("Skipping broken test; see https://github.com/ooni/probe-cli/v3/internal/engine/issues/992")
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"inputs": [
"http://www.kernel.org/",
"http://www.x.org/",
"http://www.microsoft.com/",
"http://www.slashdot.org/",
"http://www.repubblica.it/",
"http://www.google.it/",
"http://ooni.org/"
],
"name": "ExampleWithInputNonInterruptible",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
var keys []string
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
switch event.Key {
case "failure.startup":
t.Fatal(eventstr)
case "status.measurement_start":
go task.Interrupt()
}
// We compress the keys. What matters is basically that we
// see just one of the many possible measurements here.
if keys == nil || keys[len(keys)-1] != event.Key {
keys = append(keys, event.Key)
}
}
expect := []string{
"status.queued",
"status.started",
"status.progress",
"status.geoip_lookup",
"status.resolver_lookup",
"status.progress",
"status.report_create",
"status.measurement_start",
"log",
"status.progress",
"measurement",
"status.measurement_submission",
"status.measurement_done",
"status.end",
"task_terminated",
}
if diff := cmp.Diff(expect, keys); diff != "" {
t.Fatal(diff)
}
}
func TestInterruptNdt7(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"name": "Ndt7",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
go func() {
<-time.After(11 * time.Second)
task.Interrupt()
}()
var keys []string
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
if event.Key == "failure.startup" {
t.Fatal(eventstr)
}
// We compress the keys because we don't know how many
// status.progress we will see. What matters is that we
// don't see a measurement submission, since it means
// that we have interrupted the measurement.
if keys == nil || keys[len(keys)-1] != event.Key {
keys = append(keys, event.Key)
}
}
expect := []string{
"status.queued",
"status.started",
"status.progress",
"status.geoip_lookup",
"status.resolver_lookup",
"status.progress",
"status.report_create",
"status.measurement_start",
"status.progress",
"status.end",
"task_terminated",
}
if diff := cmp.Diff(expect, keys); diff != "" {
t.Fatal(diff)
}
}
func TestCountBytesForExample(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"name": "Example",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
var downloadKB, uploadKB float64
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
switch event.Key {
case "failure.startup":
t.Fatal(eventstr)
case "status.end":
downloadKB = event.Value["downloaded_kb"].(float64)
uploadKB = event.Value["uploaded_kb"].(float64)
}
}
if downloadKB == 0 {
t.Fatal("downloadKB is zero")
}
if uploadKB == 0 {
t.Fatal("uploadKB is zero")
}
}
func TestPrivacyAndScrubbing(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"name": "Example",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
var m *model.Measurement
for !task.IsDone() {
eventstr := task.WaitForNextEvent()
var event eventlike
if err := json.Unmarshal([]byte(eventstr), &event); err != nil {
t.Fatal(err)
}
switch event.Key {
case "failure.startup":
t.Fatal(eventstr)
case "measurement":
v := []byte(event.Value["json_str"].(string))
m = new(model.Measurement)
if err := json.Unmarshal(v, &m); err != nil {
t.Fatal(err)
}
}
}
if m == nil {
t.Fatal("measurement is nil")
}
if m.ProbeASN == "AS0" || m.ProbeCC == "ZZ" || m.ProbeIP != "127.0.0.1" {
t.Fatal("unexpected result")
}
}
func TestNonblock(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
task, err := oonimkall.StartTask(`{
"assets_dir": "../testdata/oonimkall/assets",
"name": "Example",
"options": {
"software_name": "oonimkall-test",
"software_version": "0.1.0"
},
"state_dir": "../testdata/oonimkall/state",
"version": 1
}`)
if err != nil {
t.Fatal(err)
}
if !task.IsRunning() {
t.Fatal("The runner should be running at this point")
}
// If the task blocks because it emits too much events, this test
// will run forever and will be killed. Because we have room for up
// to 128 events in the buffer, we should hopefully be fine.
for task.IsRunning() {
time.Sleep(time.Second)
}
for !task.IsDone() {
task.WaitForNextEvent()
}
}

5
internal/engine/oonimkall/task_internal_test.go Normal file
View file

@ -0,0 +1,5 @@
package oonimkall
func (t *Task) IsRunning() bool {
return t.isstopped.Load() == 0
}

87
internal/engine/oonimkall/tasks/chanlogger.go Normal file
View file

@ -0,0 +1,87 @@
package tasks
import (
"fmt"
)
// ChanLogger is a logger targeting a channel
type ChanLogger struct {
emitter *EventEmitter
hasdebug bool
hasinfo bool
haswarning bool
out chan<- *Event
}
// Debug implements Logger.Debug
func (cl *ChanLogger) Debug(msg string) {
if cl.hasdebug {
cl.emitter.Emit("log", EventLog{
LogLevel: "DEBUG",
Message: msg,
})
}
}
// Debugf implements Logger.Debugf
func (cl *ChanLogger) Debugf(format string, v ...interface{}) {
if cl.hasdebug {
cl.Debug(fmt.Sprintf(format, v...))
}
}
// Info implements Logger.Info
func (cl *ChanLogger) Info(msg string) {
if cl.hasinfo {
cl.emitter.Emit("log", EventLog{
LogLevel: "INFO",
Message: msg,
})
}
}
// Infof implements Logger.Infof
func (cl *ChanLogger) Infof(format string, v ...interface{}) {
if cl.hasinfo {
cl.Info(fmt.Sprintf(format, v...))
}
}
// Warn implements Logger.Warn
func (cl *ChanLogger) Warn(msg string) {
if cl.haswarning {
cl.emitter.Emit("log", EventLog{
LogLevel: "WARNING",
Message: msg,
})
}
}
// Warnf implements Logger.Warnf
func (cl *ChanLogger) Warnf(format string, v ...interface{}) {
if cl.haswarning {
cl.Warn(fmt.Sprintf(format, v...))
}
}
// NewChanLogger creates a new ChanLogger instance.
func NewChanLogger(emitter *EventEmitter, logLevel string,
out chan<- *Event) *ChanLogger {
cl := &ChanLogger{
emitter: emitter,
out: out,
}
switch logLevel {
case "DEBUG", "DEBUG2":
cl.hasdebug = true
fallthrough
case "INFO":
cl.hasinfo = true
fallthrough
case "ERR", "WARNING":
fallthrough
default:
cl.haswarning = true
}
return cl
}

57
internal/engine/oonimkall/tasks/event.go Normal file
View file

@ -0,0 +1,57 @@
package tasks
type eventEmpty struct{}
// EventFailure contains information on a failure.
type EventFailure struct {
Failure string `json:"failure"`
}
// EventLog is an event containing a log message.
type EventLog struct {
LogLevel string `json:"log_level"`
Message string `json:"message"`
}
type eventMeasurementGeneric struct {
Failure string `json:"failure,omitempty"`
Idx int64 `json:"idx"`
Input string `json:"input"`
JSONStr string `json:"json_str,omitempty"`
}
type eventStatusEnd struct {
DownloadedKB float64 `json:"downloaded_kb"`
Failure string `json:"failure"`
UploadedKB float64 `json:"uploaded_kb"`
}
type eventStatusGeoIPLookup struct {
ProbeASN string `json:"probe_asn"`
ProbeCC string `json:"probe_cc"`
ProbeIP string `json:"probe_ip"`
ProbeNetworkName string `json:"probe_network_name"`
}
// EventStatusProgress reports progress information.
type EventStatusProgress struct {
Message string `json:"message"`
Percentage float64 `json:"percentage"`
}
type eventStatusReportGeneric struct {
ReportID string `json:"report_id"`
}
type eventStatusResolverLookup struct {
ResolverASN string `json:"resolver_asn"`
ResolverIP string `json:"resolver_ip"`
ResolverNetworkName string `json:"resolver_network_name"`
}
// Event is an event emitted by a task. This structure extends the event
// described by MK v0.10.9 FFI API (https://git.io/Jv4Rv).
type Event struct {
Key string `json:"key"`
Value interface{} `json:"value"`
}

40
internal/engine/oonimkall/tasks/eventemitter.go Normal file
View file

@ -0,0 +1,40 @@
package tasks
// EventEmitter emits event on a channel
type EventEmitter struct {
disabled map[string]bool
out chan<- *Event
}
// NewEventEmitter creates a new Emitter
func NewEventEmitter(disabledEvents []string, out chan<- *Event) *EventEmitter {
ee := &EventEmitter{out: out}
ee.disabled = make(map[string]bool)
for _, eventname := range disabledEvents {
ee.disabled[eventname] = true
}
return ee
}
// EmitFailureStartup emits the failureStartup event
func (ee *EventEmitter) EmitFailureStartup(failure string) {
ee.EmitFailureGeneric(failureStartup, failure)
}
// EmitFailureGeneric emits a failure event
func (ee *EventEmitter) EmitFailureGeneric(name, failure string) {
ee.Emit(name, EventFailure{Failure: failure})
}
// EmitStatusProgress emits the status.Progress event
func (ee *EventEmitter) EmitStatusProgress(percentage float64, message string) {
ee.Emit(statusProgress, EventStatusProgress{Message: message, Percentage: percentage})
}
// Emit emits the specified event
func (ee *EventEmitter) Emit(key string, value interface{}) {
if ee.disabled[key] == true {
return
}
ee.out <- &Event{Key: key, Value: value}
}

67
internal/engine/oonimkall/tasks/eventemitter_test.go Normal file
View file

@ -0,0 +1,67 @@
package tasks_test
import (
"testing"
"github.com/ooni/probe-cli/v3/internal/engine/oonimkall/tasks"
)
func TestDisabledEvents(t *testing.T) {
out := make(chan *tasks.Event)
emitter := tasks.NewEventEmitter([]string{"log"}, out)
go func() {
emitter.Emit("log", tasks.EventLog{Message: "foo"})
close(out)
}()
var count int64
for ev := range out {
if ev.Key == "log" {
count++
}
}
if count > 0 {
t.Fatal("cannot disable events")
}
}
func TestEmitFailureStartup(t *testing.T) {
out := make(chan *tasks.Event)
emitter := tasks.NewEventEmitter([]string{}, out)
go func() {
emitter.EmitFailureStartup("mocked error")
close(out)
}()
var found bool
for ev := range out {
if ev.Key == "failure.startup" {
evv := ev.Value.(tasks.EventFailure) // panic if not castable
if evv.Failure == "mocked error" {
found = true
}
}
}
if !found {
t.Fatal("did not see expected event")
}
}
func TestEmitStatusProgress(t *testing.T) {
out := make(chan *tasks.Event)
emitter := tasks.NewEventEmitter([]string{}, out)
go func() {
emitter.EmitStatusProgress(0.7, "foo")
close(out)
}()
var found bool
for ev := range out {
if ev.Key == "status.progress" {
evv := ev.Value.(tasks.EventStatusProgress) // panic if not castable
if evv.Message == "foo" && evv.Percentage == 0.7 {
found = true
}
}
}
if !found {
t.Fatal("did not see expected event")
}
}

299
internal/engine/oonimkall/tasks/runner.go Normal file
View file

@ -0,0 +1,299 @@
// Package tasks implements tasks run using the oonimkall API.
package tasks
import (
"context"
"encoding/json"
"fmt"
"time"
engine "github.com/ooni/probe-cli/v3/internal/engine"
"github.com/ooni/probe-cli/v3/internal/engine/internal/runtimex"
"github.com/ooni/probe-cli/v3/internal/engine/model"
)
const (
failureIPLookup = "failure.ip_lookup"
failureASNLookup = "failure.asn_lookup"
failureCCLookup = "failure.cc_lookup"
failureMeasurement = "failure.measurement"
failureMeasurementSubmission = "failure.measurement_submission"
failureReportCreate = "failure.report_create"
failureResolverLookup = "failure.resolver_lookup"
failureStartup = "failure.startup"
measurement = "measurement"
statusEnd = "status.end"
statusGeoIPLookup = "status.geoip_lookup"
statusMeasurementDone = "status.measurement_done"
statusMeasurementStart = "status.measurement_start"
statusMeasurementSubmission = "status.measurement_submission"
statusProgress = "status.progress"
statusQueued = "status.queued"
statusReportCreate = "status.report_create"
statusResolverLookup = "status.resolver_lookup"
statusStarted = "status.started"
)
// Run runs the task specified by settings.Name until completion. This is the
// top-level API that should be called by oonimkall.
func Run(ctx context.Context, settings *Settings, out chan<- *Event) {
r := NewRunner(settings, out)
r.Run(ctx)
}
// Runner runs a specific task
type Runner struct {
emitter *EventEmitter
maybeLookupLocation func(*engine.Session) error
out chan<- *Event
settings *Settings
}
// NewRunner creates a new task runner
func NewRunner(settings *Settings, out chan<- *Event) *Runner {
return &Runner{
emitter: NewEventEmitter(settings.DisabledEvents, out),
out: out,
settings: settings,
}
}
// FailureInvalidVersion is the failure returned when Version is invalid
const FailureInvalidVersion = "invalid Settings.Version number"
func (r *Runner) hasUnsupportedSettings(logger *ChanLogger) bool {
if r.settings.Version < 1 {
r.emitter.EmitFailureStartup(FailureInvalidVersion)
return true
}
return false
}
func (r *Runner) newsession(logger *ChanLogger) (*engine.Session, error) {
kvstore, err := engine.NewFileSystemKVStore(r.settings.StateDir)
if err != nil {
return nil, err
}
config := engine.SessionConfig{
AssetsDir: r.settings.AssetsDir,
KVStore: kvstore,
Logger: logger,
SoftwareName: r.settings.Options.SoftwareName,
SoftwareVersion: r.settings.Options.SoftwareVersion,
TempDir: r.settings.TempDir,
}
if r.settings.Options.ProbeServicesBaseURL != "" {
config.AvailableProbeServices = []model.Service{{
Type: "https",
Address: r.settings.Options.ProbeServicesBaseURL,
}}
}
return engine.NewSession(config)
}
func (r *Runner) contextForExperiment(
ctx context.Context, builder *engine.ExperimentBuilder,
) context.Context {
if builder.Interruptible() {
return ctx
}
return context.Background()
}
type runnerCallbacks struct {
emitter *EventEmitter
}
func (cb *runnerCallbacks) OnProgress(percentage float64, message string) {
cb.emitter.Emit(statusProgress, EventStatusProgress{
Percentage: 0.4 + (percentage * 0.6), // open report is 40%
Message: message,
})
}
// Run runs the runner until completion. The context argument controls
// when to stop when processing multiple inputs, as well as when to stop
// experiments explicitly marked as interruptible.
func (r *Runner) Run(ctx context.Context) {
logger := NewChanLogger(r.emitter, r.settings.LogLevel, r.out)
r.emitter.Emit(statusQueued, eventEmpty{})
if r.hasUnsupportedSettings(logger) {
return
}
r.emitter.Emit(statusStarted, eventEmpty{})
sess, err := r.newsession(logger)
if err != nil {
r.emitter.EmitFailureStartup(err.Error())
return
}
endEvent := new(eventStatusEnd)
defer func() {
sess.Close()
r.emitter.Emit(statusEnd, endEvent)
}()
builder, err := sess.NewExperimentBuilder(r.settings.Name)
if err != nil {
r.emitter.EmitFailureStartup(err.Error())
return
}
logger.Info("Looking up OONI backends... please, be patient")
if err := sess.MaybeLookupBackends(); err != nil {
r.emitter.EmitFailureStartup(err.Error())
return
}
r.emitter.EmitStatusProgress(0.1, "contacted bouncer")
logger.Info("Looking up your location... please, be patient")
maybeLookupLocation := r.maybeLookupLocation
if maybeLookupLocation == nil {
maybeLookupLocation = func(sess *engine.Session) error {
return sess.MaybeLookupLocation()
}
}
if err := maybeLookupLocation(sess); err != nil {
r.emitter.EmitFailureGeneric(failureIPLookup, err.Error())
r.emitter.EmitFailureGeneric(failureASNLookup, err.Error())
r.emitter.EmitFailureGeneric(failureCCLookup, err.Error())
r.emitter.EmitFailureGeneric(failureResolverLookup, err.Error())
return
}
r.emitter.EmitStatusProgress(0.2, "geoip lookup")
r.emitter.EmitStatusProgress(0.3, "resolver lookup")
r.emitter.Emit(statusGeoIPLookup, eventStatusGeoIPLookup{
ProbeIP: sess.ProbeIP(),
ProbeASN: sess.ProbeASNString(),
ProbeCC: sess.ProbeCC(),
ProbeNetworkName: sess.ProbeNetworkName(),
})
r.emitter.Emit(statusResolverLookup, eventStatusResolverLookup{
ResolverASN: sess.ResolverASNString(),
ResolverIP: sess.ResolverIP(),
ResolverNetworkName: sess.ResolverNetworkName(),
})
builder.SetCallbacks(&runnerCallbacks{emitter: r.emitter})
if len(r.settings.Inputs) <= 0 {
switch builder.InputPolicy() {
case engine.InputOrQueryTestLists, engine.InputStrictlyRequired:
r.emitter.EmitFailureStartup("no input provided")
return
}
r.settings.Inputs = append(r.settings.Inputs, "")
}
experiment := builder.NewExperiment()
defer func() {
endEvent.DownloadedKB = experiment.KibiBytesReceived()
endEvent.UploadedKB = experiment.KibiBytesSent()
}()
if !r.settings.Options.NoCollector {
logger.Info("Opening report... please, be patient")
if err := experiment.OpenReport(); err != nil {
r.emitter.EmitFailureGeneric(failureReportCreate, err.Error())
return
}
r.emitter.EmitStatusProgress(0.4, "open report")
r.emitter.Emit(statusReportCreate, eventStatusReportGeneric{
ReportID: experiment.ReportID(),
})
}
// This deviates a little bit from measurement-kit, for which
// a zero timeout is actually valid. Since it does not make much
// sense, here we're changing the behaviour.
//
// See https://github.com/measurement-kit/measurement-kit/issues/1922
if r.settings.Options.MaxRuntime > 0 {
// We want to honour max_runtime only when we're running an
// experiment that clearly wants specific input. We could refine
// this policy in the future, but for now this covers in a
// reasonable way web connectivity, so we should be ok.
switch builder.InputPolicy() {
case engine.InputOrQueryTestLists, engine.InputStrictlyRequired:
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(
ctx, time.Duration(r.settings.Options.MaxRuntime)*time.Second,
)
defer cancel()
}
}
inputCount := len(r.settings.Inputs)
start := time.Now()
inflatedMaxRuntime := r.settings.Options.MaxRuntime + r.settings.Options.MaxRuntime/10
eta := start.Add(time.Duration(inflatedMaxRuntime) * time.Second)
for idx, input := range r.settings.Inputs {
if ctx.Err() != nil {
break
}
logger.Infof("Starting measurement with index %d", idx)
r.emitter.Emit(statusMeasurementStart, eventMeasurementGeneric{
Idx: int64(idx),
Input: input,
})
if input != "" && inputCount > 0 {
var percentage float64
if r.settings.Options.MaxRuntime > 0 {
now := time.Now()
percentage = (now.Sub(start).Seconds()/eta.Sub(start).Seconds())*0.6 + 0.4
} else {
percentage = (float64(idx)/float64(inputCount))*0.6 + 0.4
}
r.emitter.EmitStatusProgress(percentage, fmt.Sprintf(
"processing %s", input,
))
}
m, err := experiment.MeasureWithContext(
r.contextForExperiment(ctx, builder),
input,
)
if builder.Interruptible() && ctx.Err() != nil {
// We want to stop here only if interruptible otherwise we want to
// submit measurement and stop at beginning of next iteration
break
}
m.AddAnnotations(r.settings.Annotations)
if err != nil {
r.emitter.Emit(failureMeasurement, eventMeasurementGeneric{
Failure: err.Error(),
Idx: int64(idx),
Input: input,
})
// fallthrough: we want to submit the report anyway
}
data, err := json.Marshal(m)
runtimex.PanicOnError(err, "measurement.MarshalJSON failed")
r.emitter.Emit(measurement, eventMeasurementGeneric{
Idx: int64(idx),
Input: input,
JSONStr: string(data),
})
if !r.settings.Options.NoCollector {
logger.Info("Submitting measurement... please, be patient")
err := experiment.SubmitAndUpdateMeasurement(m)
r.emitter.Emit(measurementSubmissionEventName(err), eventMeasurementGeneric{
Idx: int64(idx),
Input: input,
JSONStr: string(data),
Failure: measurementSubmissionFailure(err),
})
}
r.emitter.Emit(statusMeasurementDone, eventMeasurementGeneric{
Idx: int64(idx),
Input: input,
})
}
}
func measurementSubmissionEventName(err error) string {
if err != nil {
return failureMeasurementSubmission
}
return statusMeasurementSubmission
}
func measurementSubmissionFailure(err error) string {
if err != nil {
return err.Error()
}
return ""
}

404
internal/engine/oonimkall/tasks/runner_integration_test.go Normal file
View file

@ -0,0 +1,404 @@
package tasks_test
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"github.com/ooni/probe-cli/v3/internal/engine/oonimkall/tasks"
)
func TestRunnerMaybeLookupBackendsFailure(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(500)
}))
defer server.Close()
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
Name: "Example",
Options: tasks.SettingsOptions{
ProbeServicesBaseURL: server.URL,
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
Version: 1,
}
go func() {
tasks.Run(context.Background(), settings, out)
close(out)
}()
var failures []string
for ev := range out {
switch ev.Key {
case "failure.startup":
failure := ev.Value.(tasks.EventFailure).Failure
failures = append(failures, failure)
case "status.queued", "status.started", "log", "status.end":
default:
panic(fmt.Sprintf("unexpected key: %s", ev.Key))
}
}
if len(failures) != 1 {
t.Fatal("unexpected number of failures")
}
if failures[0] != "all available probe services failed" {
t.Fatal("invalid failure")
}
}
func TestRunnerOpenReportFailure(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
var (
nreq int64
mu sync.Mutex
)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
nreq++
if nreq == 1 {
w.Write([]byte(`{}`))
return
}
w.WriteHeader(500)
}))
defer server.Close()
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
Name: "Example",
Options: tasks.SettingsOptions{
ProbeServicesBaseURL: server.URL,
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
Version: 1,
}
seench := make(chan int64)
go func() {
var seen int64
for ev := range out {
switch ev.Key {
case "failure.report_create":
seen++
case "status.progress":
evv := ev.Value.(tasks.EventStatusProgress)
if evv.Percentage >= 0.4 {
panic(fmt.Sprintf("too much progress: %+v", ev))
}
case "status.queued", "status.started", "log", "status.end",
"status.geoip_lookup", "status.resolver_lookup":
default:
panic(fmt.Sprintf("unexpected key: %s", ev.Key))
}
}
seench <- seen
}()
tasks.Run(context.Background(), settings, out)
close(out)
if n := <-seench; n != 1 {
t.Fatal("unexpected number of events")
}
}
func TestRunnerGood(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
LogLevel: "DEBUG",
Name: "Example",
Options: tasks.SettingsOptions{
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
Version: 1,
}
go func() {
tasks.Run(context.Background(), settings, out)
close(out)
}()
var found bool
for ev := range out {
if ev.Key == "status.end" {
found = true
}
}
if !found {
t.Fatal("status.end event not found")
}
}
func TestRunnerWithUnsupportedSettings(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
LogLevel: "DEBUG",
Name: "Example",
Options: tasks.SettingsOptions{
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
}
go func() {
tasks.Run(context.Background(), settings, out)
close(out)
}()
var failures []string
for ev := range out {
if ev.Key == "failure.startup" {
failure := ev.Value.(tasks.EventFailure).Failure
failures = append(failures, failure)
}
}
if len(failures) != 1 {
t.Fatal("invalid number of failures")
}
if failures[0] != tasks.FailureInvalidVersion {
t.Fatal("not the failure we expected")
}
}
func TestRunnerWithInvalidKVStorePath(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
LogLevel: "DEBUG",
Name: "Example",
Options: tasks.SettingsOptions{
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "", // must be empty to cause the failure below
Version: 1,
}
go func() {
tasks.Run(context.Background(), settings, out)
close(out)
}()
var failures []string
for ev := range out {
if ev.Key == "failure.startup" {
failure := ev.Value.(tasks.EventFailure).Failure
failures = append(failures, failure)
}
}
if len(failures) != 1 {
t.Fatal("invalid number of failures")
}
if failures[0] != "mkdir : no such file or directory" {
t.Fatal("not the failure we expected")
}
}
func TestRunnerWithInvalidExperimentName(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
LogLevel: "DEBUG",
Name: "Nonexistent",
Options: tasks.SettingsOptions{
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
Version: 1,
}
go func() {
tasks.Run(context.Background(), settings, out)
close(out)
}()
var failures []string
for ev := range out {
if ev.Key == "failure.startup" {
failure := ev.Value.(tasks.EventFailure).Failure
failures = append(failures, failure)
}
}
if len(failures) != 1 {
t.Fatal("invalid number of failures")
}
if failures[0] != "no such experiment: Nonexistent" {
t.Fatalf("not the failure we expected: %s", failures[0])
}
}
func TestRunnerWithMissingInput(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
LogLevel: "DEBUG",
Name: "ExampleWithInput",
Options: tasks.SettingsOptions{
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
Version: 1,
}
go func() {
tasks.Run(context.Background(), settings, out)
close(out)
}()
var failures []string
for ev := range out {
if ev.Key == "failure.startup" {
failure := ev.Value.(tasks.EventFailure).Failure
failures = append(failures, failure)
}
}
if len(failures) != 1 {
t.Fatal("invalid number of failures")
}
if failures[0] != "no input provided" {
t.Fatalf("not the failure we expected: %s", failures[0])
}
}
func TestRunnerWithMaxRuntime(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
Inputs: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i"},
LogLevel: "DEBUG",
Name: "ExampleWithInput",
Options: tasks.SettingsOptions{
MaxRuntime: 1,
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
Version: 1,
}
begin := time.Now()
go func() {
tasks.Run(context.Background(), settings, out)
close(out)
}()
var found bool
for ev := range out {
if ev.Key == "status.end" {
found = true
}
}
if !found {
t.Fatal("status.end event not found")
}
// The runtime is long because of ancillary operations and is even more
// longer because of self shaping we may be performing (especially in
// CI builds) using `-tags shaping`). We have experimentally determined
// that ~10 seconds is the typical CI test run time. See:
//
// 1. https://github.com/ooni/probe-cli/v3/internal/engine/pull/588/checks?check_run_id=667263788
//
// 2. https://github.com/ooni/probe-cli/v3/internal/engine/pull/588/checks?check_run_id=667263855
if time.Now().Sub(begin) > 10*time.Second {
t.Fatal("expected shorter runtime")
}
}
func TestRunnerWithMaxRuntimeNonInterruptible(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
Inputs: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i"},
LogLevel: "DEBUG",
Name: "ExampleWithInputNonInterruptible",
Options: tasks.SettingsOptions{
MaxRuntime: 1,
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
Version: 1,
}
begin := time.Now()
go func() {
tasks.Run(context.Background(), settings, out)
close(out)
}()
var found bool
for ev := range out {
if ev.Key == "status.end" {
found = true
}
}
if !found {
t.Fatal("status.end event not found")
}
// The runtime is long because of ancillary operations and is even more
// longer because of self shaping we may be performing (especially in
// CI builds) using `-tags shaping`). We have experimentally determined
// that ~10 seconds is the typical CI test run time. See:
//
// 1. https://github.com/ooni/probe-cli/v3/internal/engine/pull/588/checks?check_run_id=667263788
//
// 2. https://github.com/ooni/probe-cli/v3/internal/engine/pull/588/checks?check_run_id=667263855
if time.Now().Sub(begin) > 10*time.Second {
t.Fatal("expected shorter runtime")
}
}
func TestRunnerWithFailedMeasurement(t *testing.T) {
if testing.Short() {
t.Skip("skip test in short mode")
}
out := make(chan *tasks.Event)
settings := &tasks.Settings{
AssetsDir: "../../testdata/oonimkall/assets",
Inputs: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i"},
LogLevel: "DEBUG",
Name: "ExampleWithFailure",
Options: tasks.SettingsOptions{
MaxRuntime: 1,
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
Version: 1,
}
go func() {
tasks.Run(context.Background(), settings, out)
close(out)
}()
var found bool
for ev := range out {
if ev.Key == "failure.measurement" {
found = true
}
}
if !found {
t.Fatal("failure.measurement event not found")
}
}

72
internal/engine/oonimkall/tasks/runner_internal_test.go Normal file
View file

@ -0,0 +1,72 @@
package tasks
import (
"context"
"errors"
"fmt"
"testing"
engine "github.com/ooni/probe-cli/v3/internal/engine"
)
func TestMeasurementSubmissionEventName(t *testing.T) {
if measurementSubmissionEventName(nil) != statusMeasurementSubmission {
t.Fatal("unexpected submission event name")
}
if measurementSubmissionEventName(errors.New("mocked error")) != failureMeasurementSubmission {
t.Fatal("unexpected submission event name")
}
}
func TestMeasurementSubmissionFailure(t *testing.T) {
if measurementSubmissionFailure(nil) != "" {
t.Fatal("unexpected submission failure")
}
if measurementSubmissionFailure(errors.New("mocked error")) != "mocked error" {
t.Fatal("unexpected submission failure")
}
}
func TestRunnerMaybeLookupLocationFailure(t *testing.T) {
out := make(chan *Event)
settings := &Settings{
AssetsDir: "../../testdata/oonimkall/assets",
Name: "Example",
Options: SettingsOptions{
SoftwareName: "oonimkall-test",
SoftwareVersion: "0.1.0",
},
StateDir: "../../testdata/oonimkall/state",
Version: 1,
}
seench := make(chan int64)
go func() {
var seen int64
for ev := range out {
switch ev.Key {
case "failure.ip_lookup", "failure.asn_lookup",
"failure.cc_lookup", "failure.resolver_lookup":
seen++
case "status.progress":
evv := ev.Value.(EventStatusProgress)
if evv.Percentage >= 0.2 {
panic(fmt.Sprintf("too much progress: %+v", ev))
}
case "status.queued", "status.started", "status.end":
default:
panic(fmt.Sprintf("unexpected key: %s", ev.Key))
}
}
seench <- seen
}()
expected := errors.New("mocked error")
r := NewRunner(settings, out)
r.maybeLookupLocation = func(*engine.Session) error {
return expected
}
r.Run(context.Background())
close(out)
if n := <-seench; n != 4 {
t.Fatal("unexpected number of events")
}
}

73
internal/engine/oonimkall/tasks/settings.go Normal file
View file

@ -0,0 +1,73 @@
package tasks
// Settings contains settings for a task. This structure derives from
// the one described by MK v0.10.9 FFI API (https://git.io/Jv4Rv), yet
// since 2020-12-03 we're not backwards compatible anymore.
type Settings struct {
// Annotations contains the annotations to be added
// to every measurements performed by the task.
Annotations map[string]string `json:"annotations,omitempty"`
// AssetsDir is the directory where to store assets. This
// field is an extension of MK's specification. If
// this field is empty, the task won't start.
AssetsDir string `json:"assets_dir"`
// DisabledEvents contains disabled events. See
// https://git.io/Jv4Rv for the events names.
DisabledEvents []string `json:"disabled_events,omitempty"`
// Inputs contains the inputs. The task will fail if it
// requires input and you provide no input.
Inputs []string `json:"inputs,omitempty"`
// LogLevel contains the logs level. See https://git.io/Jv4Rv
// for the names of the available log levels.
LogLevel string `json:"log_level,omitempty"`
// Name contains the task name. By https://git.io/Jv4Rv the
// names are in camel case, e.g. `Ndt`.
Name string `json:"name"`
// Options contains the task options.
Options SettingsOptions `json:"options"`
// StateDir is the directory where to store persistent data. This
// field is an extension of MK's specification. If
// this field is empty, the task won't start.
StateDir string `json:"state_dir"`
// TempDir is the temporary directory. This field is an extension of MK's
// specification. If this field is empty, we will pick the tempdir that
// ioutil.TempDir uses by default, which may not work on mobile. According
// to our experiments as of 2020-06-10, leaving the TempDir empty works
// for iOS and does not work for Android.
TempDir string `json:"temp_dir"`
// Version indicates the version of this structure.
Version int64 `json:"version"`
}
// SettingsOptions contains the settings options
type SettingsOptions struct {
// MaxRuntime is the maximum runtime expressed in seconds. A negative
// value for this field disables the maximum runtime. Using
// a zero value will also mean disabled. This is not the
// original behaviour of Measurement Kit, which used to run
// for zero time in such case.
MaxRuntime float64 `json:"max_runtime,omitempty"`
// NoCollector indicates whether to use a collector
NoCollector bool `json:"no_collector,omitempty"`
// ProbeServicesBaseURL contains the probe services base URL.
ProbeServicesBaseURL string `json:"probe_services_base_url,omitempty"`
// SoftwareName is the software name. If this option is not
// present, then the library startup will fail.
SoftwareName string `json:"software_name,omitempty"`
// SoftwareVersion is the software version. If this option is not
// present, then the library startup will fail.
SoftwareVersion string `json:"software_version,omitempty"`
}

9
internal/engine/oonimkall/uuid.go Normal file
View file

@ -0,0 +1,9 @@
package oonimkall
import "github.com/google/uuid"
// NewUUID4 generates a new UUID4 string. This functionality is typically
// used by mobile apps to generate random unique identifiers.
func NewUUID4() string {
return uuid.Must(uuid.NewRandom()).String()
}

13
internal/engine/oonimkall/uuid_test.go Normal file
View file

@ -0,0 +1,13 @@
package oonimkall_test
import (
"testing"
"github.com/ooni/probe-cli/v3/internal/engine/oonimkall"
)
func TestNewUUID4(t *testing.T) {
if out := oonimkall.NewUUID4(); len(out) != 36 {
t.Fatal("not the expected output")
}
}